Repository: knative-sandbox/eventing-kafka-broker
Pull request #1819 (2022-01-27 14:57:47)

Add scheduler and de-scheduler configuration
- Use the same configurations used in eventing-kafka

Part of #1537

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

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

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

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

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

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

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1819/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1819?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff            @@
##               main    #1819   +/-   ##
=========================================
  Coverage     68.79%   68.79%           
  Complexity      614      614           
=========================================
  Files           131      131           
  Lines          5492     5492           
  Branches        178      178           
=========================================
  Hits           3778     3778           
  Misses         1410     1410           
  Partials        304      304           
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.01% <ø> (ø)` | |

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


------

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

knative-prow-robot: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

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

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

<details>

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

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
pierDipi: /test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests

Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index ec5cec4b4..600ae4aac 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -170,6 +170,7 @@ aliases:
   - nicolaferraro
   - rhuss
   knative-admin:
+  - carlisia
   - csantanapr
   - dprotaso
   - evankanderson
@@ -180,13 +181,13 @@ aliases:
   - knative-test-reporter-robot
   - lance
   - pmorie
-  - psschwei
   - rhuss
   - thisisnotapril
   - vaikas
+  - xtreme-sameer-vohra
   knative-release-leads:
-  - dprotaso
-  - psschwei
+  - carlisia
+  - xtreme-sameer-vohra
   knative-robots:
   - knative-prow-releaser-robot
   - knative-prow-robot
@@ -230,7 +231,9 @@ aliases:
   productivity-wg-leads:
   - chizhg
   productivity-writers:
+  - cardil
   - chizhg
+  - psschwei
   security-wg-leads:
   - evankanderson
   - julz
diff --git a/control-plane/config/200-controller/100-config-kafka-descheduler.yaml b/control-plane/config/200-controller/100-config-kafka-descheduler.yaml
new file mode 100644
index 000000000..00b90633a
--- /dev/null
+++ b/control-plane/config/200-controller/100-config-kafka-descheduler.yaml
@@ -0,0 +1,61 @@
+# 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
+#
+#     https://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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-descheduler
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+data:
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+    #
+    # These predicate plugins are used to filter out pods that
+    # a vreplica cannot be removed from due to not satisfying the
+    # predicate condition. For each pod, the descheduler
+    # will call filter plugins in their configured order.
+    # predicates: |+
+                    []
+    #
+    # These priority plugins are used to score each pod that
+    # has passed the filtering phase. The scheduler will then
+    # select the pod with the highest weighted score sum to
+    # remove vreplica from. If two pods have an equal score,
+    # a pod is chosen randomly.
+    # priorities: |+
+                  [
+                    {"Name": "RemoveWithEvenPodSpreadPriority",
+                    "Weight": 10,
+                    "Args": "{\"MaxSkew\": 2}"},
+                    {"Name": "RemoveWithAvailabilityZonePriority",
+                    "Weight": 10,
+                    "Args":  "{\"MaxSkew\": 2}"},
+                    {"Name": "RemoveWithHighestOrdinalPriority",
+                    "Weight": 2}
+                  ]
diff --git a/control-plane/config/200-controller/100-config-kafka-scheduler.yaml b/control-plane/config/200-controller/100-config-kafka-scheduler.yaml
new file mode 100644
index 000000000..a419b8471
--- /dev/null
+++ b/control-plane/config/200-controller/100-config-kafka-scheduler.yaml
@@ -0,0 +1,63 @@
+# 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
+#
+#     https://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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-scheduler
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+data:
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+    #
+    # These predicate plugins are used to filter out pods that
+    # a vreplica cannot be placed on due to not satisfying the
+    # predicate condition. For each pod, the scheduler
+    # will call filter plugins in their configured order.
+    # predicates: |+
+                  [
+                    {"Name": "PodFitsResources"},
+                    {"Name": "NoMaxResourceCount",
+                    "Args": "{\"NumPartitions\": 100}"},
+                    {"Name": "EvenPodSpread",
+                    "Args": "{\"MaxSkew\": 2}"}
+                  ]
+    #
+    # These priority plugins are used to score each pod that
+    # has passed the filtering phase. The scheduler will then
+    # select the pod with the highest weighted score sum. If
+    # two pods have an equal score, a pod is chosen randomly.
+    # priorities: |+
+                  [
+                    {"Name": "AvailabilityZonePriority",
+                    "Weight": 10,
+                    "Args":  "{\"MaxSkew\": 2}"},
+                    {"Name": "LowestOrdinalPriority",
+                    "Weight": 2}
+                  ]
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index df9794c36..1eb85a29b 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -219,11 +219,6 @@ func (manager *StatusConditionManager) FailedToUpdateReceiverPodsAnnotation(err
 	return fmt.Errorf("failed to update receiver pods annotation: %w", err)
 }
 
-func (manager *StatusConditionManager) FailedToGetConfig(err error) reconciler.Event {
-
-	return fmt.Errorf("failed to get contract configuration: %w", err)
-}
-
 func (manager *StatusConditionManager) FailedToResolveConfig(err error) reconciler.Event {
 
 	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 08ea153b0..c2bdb9fa9 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -161,7 +161,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	// Get resource configuration.
 	brokerResource, err := r.reconcilerBrokerResource(ctx, topic, broker, secret, topicConfig)
 	if err != nil {
-		return statusConditionManager.FailedToGetConfig(err)
+		return statusConditionManager.FailedToResolveConfig(err)
 	}
 	coreconfig.SetDeadLetterSinkURIFromEgressConfig(&broker.Status.DeliveryStatus, brokerResource.EgressConfig)
 
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index c3a6be022..93a874d70 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -18,15 +18,13 @@ package channel
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strconv"
 
 	"github.com/Shopify/sarama"
+
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -39,7 +37,6 @@ import (
 	"knative.dev/pkg/system"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
-	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/constants"
@@ -80,8 +77,7 @@ type Reconciler struct {
 	// reconciliation loop.
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
-	ConfigMapLister    corelisters.ConfigMapLister
-	SubscriptionLister messaginglisters.SubscriptionLister
+	ConfigMapLister corelisters.ConfigMapLister
 
 	Prober prober.Prober
 
@@ -188,7 +184,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	// Get data plane config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
-		return statusConditionManager.FailedToGetConfig(err)
+		return statusConditionManager.FailedToResolveConfig(err)
 	}
 	logger.Debug("Got contract config map")
 
@@ -202,7 +198,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	// Get resource configuration
 	channelResource, err := r.getChannelContractResource(ctx, topic, channel, secret, topicConfig)
 	if err != nil {
-		return statusConditionManager.FailedToGetConfig(err)
+		return statusConditionManager.FailedToResolveConfig(err)
 	}
 	coreconfig.SetDeadLetterSinkURIFromEgressConfig(&channel.Status.DeliveryStatus, channelResource.EgressConfig)
 
@@ -480,21 +476,11 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 }
 
 func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
-	subscriptionName, err := r.getSubscriptionName(channel, subscriber)
-	if err != nil {
-		return nil, fmt.Errorf("failed to extract subscription name for subscriber %v: %w", subscriber, err)
-	}
-
 	egress := &contract.Egress{
 		Destination:   subscriber.SubscriberURI.String(),
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
-		Reference: &contract.Reference{
-			Uuid:      string(subscriber.UID),
-			Namespace: channel.GetNamespace(),
-			Name:      subscriptionName,
-		},
 		ReplyStrategy: &contract.Egress_DiscardReply{},
 	}
 
@@ -595,21 +581,6 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
-func (r *Reconciler) getSubscriptionName(channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (string, error) {
-	subscriptions, err := r.SubscriptionLister.Subscriptions(channel.GetNamespace()).List(labels.Everything())
-	if err != nil {
-		return "", fmt.Errorf("failed to list subscriptions in namespace %s: %w", channel.GetNamespace(), err)
-	}
-
-	for _, s := range subscriptions {
-		if s.UID == subscriber.UID {
-			return s.Name, nil
-		}
-	}
-
-	return "", errors.New(string(metav1.StatusReasonNotFound))
-}
-
 // consumerGroup returns a consumerGroup name for the given channel and subscription
 func consumerGroup(channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec) string {
 	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(sub.UID))
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index db8296757..8ccfa800b 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -25,8 +25,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 
-	subscriptioninformer "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
-
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
 	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
@@ -67,7 +65,6 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
-		SubscriptionLister:         subscriptioninformer.Get(ctx).Lister(),
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		InitOffsetsFunc:            offset.InitOffsets,
diff --git a/control-plane/pkg/reconciler/consumer/controller.go b/control-plane/pkg/reconciler/consumer/controller.go
new file mode 100644
index 000000000..7b7882fec
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumer/controller.go
@@ -0,0 +1,66 @@
+/*
+ * 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 consumer
+
+import (
+	"context"
+
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/resolver"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
+	creconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumer"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+)
+
+func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
+
+	consumerInformer := consumer.Get(ctx)
+
+	r := &Reconciler{
+		SerDe:               formatSerDeFromString(configs.DataPlaneConfigFormat),
+		ConsumerGroupLister: consumergroup.Get(ctx).Lister(),
+		SecretLister:        secretinformer.Get(ctx).Lister(),
+		PodLister:           podinformer.Get(ctx).Lister(),
+		KubeClient:          kubeclient.Get(ctx),
+	}
+
+	impl := creconciler.NewImpl(ctx, r)
+
+	r.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
+
+	consumerInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+
+	r.Tracker = impl.Tracker
+	secretinformer.Get(ctx).Informer().AddEventHandler(controller.HandleAll(r.Tracker.OnChanged))
+
+	return impl
+}
+
+func formatSerDeFromString(val string) contract.FormatSerDe {
+	switch val {
+	case "protobuf":
+		return contract.FormatSerDe{Format: contract.Protobuf}
+	default:
+		return contract.FormatSerDe{Format: contract.Json}
+	}
+}
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 1aaabe7c3..e732754de 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -154,7 +154,7 @@ Lists of 158 third-party dependencies.
      (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.13 - https://www.scala-lang.org/)
      (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.3.0 - http://www.scala-lang.org/)
      (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 - http://www.scala-lang.org/)
-     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.33 - http://www.slf4j.org)
-     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.33 - http://www.slf4j.org)
+     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.35 - http://www.slf4j.org)
+     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.35 - http://www.slf4j.org)
      (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.8.1 - https://github.com/xerial/snappy-java)
      (Apache License, Version 2.0) SnakeYAML (org.yaml:snakeyaml:1.28 - http://www.snakeyaml.org)
diff --git a/data-plane/config/sourcev2/500-dispatcher.yaml b/data-plane/config/sourcev2/500-dispatcher.yaml
new file mode 100644
index 000000000..770525ad2
--- /dev/null
+++ b/data-plane/config/sourcev2/500-dispatcher.yaml
@@ -0,0 +1,157 @@
+---
+
+# 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.
+
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: kafka-source-dispatcher
+  namespace: knative-eventing
+  labels:
+    app: kafka-source-dispatcher
+    kafka.eventing.knative.dev/release: devel
+spec:
+  serviceName: kafka-source-dispatcher
+  selector:
+    matchLabels:
+      app: kafka-source-dispatcher
+  template:
+    metadata:
+      name: kafka-source-dispatcher
+      labels:
+        app: kafka-source-dispatcher
+        kafka.eventing.knative.dev/release: devel
+    spec:
+      serviceAccountName: knative-kafka-source-data-plane
+      securityContext:
+        runAsNonRoot: true
+      containers:
+        - name: kafka-source-dispatcher
+          image: ${KNATIVE_KAFKA_DISPATCHER_IMAGE}
+          imagePullPolicy: IfNotPresent
+          volumeMounts:
+            - mountPath: /etc/config
+              name: config-kafka-source-data-plane
+              readOnly: true
+            - mountPath: /etc/contract-resources
+              name: contract-resources
+              readOnly: true
+            - mountPath: /tmp
+              name: cache
+            - mountPath: /etc/logging
+              name: kafka-config-logging
+              readOnly: true
+            - mountPath: /etc/tracing
+              name: config-tracing
+              readOnly: true
+          ports:
+            - containerPort: 9090
+              name: http-metrics
+              protocol: TCP
+          env:
+            - name: SERVICE_NAME
+              value: "kafka-source-dispatcher"
+            - name: SERVICE_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+            - name: PRODUCER_CONFIG_FILE_PATH
+              value: /etc/config/config-kafka-source-producer.properties
+            - name: CONSUMER_CONFIG_FILE_PATH
+              value: /etc/config/config-kafka-source-consumer.properties
+            - name: WEBCLIENT_CONFIG_FILE_PATH
+              value: /etc/config/config-kafka-source-webclient.properties
+            - name: DATA_PLANE_CONFIG_FILE_PATH
+              value: /etc/contract-resources/data
+            - name: EGRESSES_INITIAL_CAPACITY
+              value: "20"
+            - name: INSTANCE_ID
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.uid
+            - name: METRICS_PATH
+              value: /metrics
+            - name: METRICS_PORT
+              value: "9090"
+            - name: METRICS_PUBLISH_QUANTILES
+              value: "false"
+            - name: METRICS_JVM_ENABLED
+              value: "false"
+            - name: CONFIG_TRACING_PATH
+              value: "/etc/tracing"
+            # https://github.com/fabric8io/kubernetes-client/issues/2212
+            - name: HTTP2_DISABLE
+              value: "true"
+            # This should be set according to initial delay seconds
+            - name: WAIT_STARTUP_SECONDS
+              value: "8"
+          command:
+            - "java"
+          args:
+            - "-Dlogback.configurationFile=/etc/logging/config.xml"
+            - "-jar"
+            - "/app/app.jar"
+
+          resources:
+            requests:
+              cpu: 1000m
+              # 600Mi for virtual replicas + 100Mi overhead
+              memory: 700Mi
+            limits:
+              cpu: 2000m
+              memory: 1000Mi
+
+          livenessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 10
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
+          readinessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 10
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
+          terminationMessagePolicy: FallbackToLogsOnError
+          terminationMessagePath: /dev/temination-log
+          securityContext:
+            allowPrivilegeEscalation: false
+            privileged: false
+            readOnlyRootFilesystem: true
+      volumes:
+        - name: config-kafka-source-data-plane
+          configMap:
+            name: config-kafka-source-data-plane
+        - name: cache
+          emptyDir: { }
+        - name: kafka-config-logging
+          configMap:
+            name: kafka-config-logging
+        - name: config-tracing
+          configMap:
+            name: config-tracing
+      restartPolicy: Always
+      dnsConfig:
+        options:
+          - name: single-request-reopen
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index eaddaf405..910aa857a 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -47,7 +47,7 @@
     <opentelemetry.version>1.10.1</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
     <protobuf.version>3.19.3</protobuf.version>
-    <slf4j.version>1.7.33</slf4j.version>
+    <slf4j.version>1.7.35</slf4j.version>
     <ch.qos.logback.version>1.2.10</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
     <assertj.version>3.22.0</assertj.version>
diff --git a/hack/release.sh b/hack/release.sh
index a7a7c295f..e6a82d0ac 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -67,6 +67,11 @@ function build_release() {
     "${EVENTING_KAFKA_CHANNEL_ARTIFACT}"
     "${EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT}"
   )
+
+  # ARTIFACTS_TO_PUBLISH has to be a string, not an array.
+  # shellcheck disable=SC2178
+  # shellcheck disable=SC2124
+  export ARTIFACTS_TO_PUBLISH="${ARTIFACTS_TO_PUBLISH[@]}"
 }
 
 main $@
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
deleted file mode 100644
index eb18a1fac..000000000
--- a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Code generated by injection-gen. DO NOT EDIT.
-
-package fake
-
-import (
-	context "context"
-
-	fake "knative.dev/eventing/pkg/client/injection/informers/factory/fake"
-	subscription "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
-	controller "knative.dev/pkg/controller"
-	injection "knative.dev/pkg/injection"
-)
-
-var Get = subscription.Get
-
-func init() {
-	injection.Fake.RegisterInformer(withInformer)
-}
-
-func withInformer(ctx context.Context) (context.Context, controller.Informer) {
-	f := fake.Get(ctx)
-	inf := f.Messaging().V1().Subscriptions()
-	return context.WithValue(ctx, subscription.Key{}, inf), inf.Informer()
-}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
deleted file mode 100644
index 37862d8a5..000000000
--- a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Code generated by injection-gen. DO NOT EDIT.
-
-package subscription
-
-import (
-	context "context"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	labels "k8s.io/apimachinery/pkg/labels"
-	cache "k8s.io/client-go/tools/cache"
-	apismessagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
-	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
-	v1 "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
-	client "knative.dev/eventing/pkg/client/injection/client"
-	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
-	messagingv1 "knative.dev/eventing/pkg/client/listers/messaging/v1"
-	controller "knative.dev/pkg/controller"
-	injection "knative.dev/pkg/injection"
-	logging "knative.dev/pkg/logging"
-)
-
-func init() {
-	injection.Default.RegisterInformer(withInformer)
-	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
-}
-
-// Key is used for associating the Informer inside the context.Context.
-type Key struct{}
-
-func withInformer(ctx context.Context) (context.Context, controller.Informer) {
-	f := factory.Get(ctx)
-	inf := f.Messaging().V1().Subscriptions()
-	return context.WithValue(ctx, Key{}, inf), inf.Informer()
-}
-
-func withDynamicInformer(ctx context.Context) context.Context {
-	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
-	return context.WithValue(ctx, Key{}, inf)
-}
-
-// Get extracts the typed informer from the context.
-func Get(ctx context.Context) v1.SubscriptionInformer {
-	untyped := ctx.Value(Key{})
-	if untyped == nil {
-		logging.FromContext(ctx).Panic(
-			"Unable to fetch knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1.SubscriptionInformer from context.")
-	}
-	return untyped.(v1.SubscriptionInformer)
-}
-
-type wrapper struct {
-	client versioned.Interface
-
-	namespace string
-
-	resourceVersion string
-}
-
-var _ v1.SubscriptionInformer = (*wrapper)(nil)
-var _ messagingv1.SubscriptionLister = (*wrapper)(nil)
-
-func (w *wrapper) Informer() cache.SharedIndexInformer {
-	return cache.NewSharedIndexInformer(nil, &apismessagingv1.Subscription{}, 0, nil)
-}
-
-func (w *wrapper) Lister() messagingv1.SubscriptionLister {
-	return w
-}
-
-func (w *wrapper) Subscriptions(namespace string) messagingv1.SubscriptionNamespaceLister {
-	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
-}
-
-// SetResourceVersion allows consumers to adjust the minimum resourceVersion
-// used by the underlying client.  It is not accessible via the standard
-// lister interface, but can be accessed through a user-defined interface and
-// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
-func (w *wrapper) SetResourceVersion(resourceVersion string) {
-	w.resourceVersion = resourceVersion
-}
-
-func (w *wrapper) List(selector labels.Selector) (ret []*apismessagingv1.Subscription, err error) {
-	lo, err := w.client.MessagingV1().Subscriptions(w.namespace).List(context.TODO(), metav1.ListOptions{
-		LabelSelector:   selector.String(),
-		ResourceVersion: w.resourceVersion,
-	})
-	if err != nil {
-		return nil, err
-	}
-	for idx := range lo.Items {
-		ret = append(ret, &lo.Items[idx])
-	}
-	return ret, nil
-}
-
-func (w *wrapper) Get(name string) (*apismessagingv1.Subscription, error) {
-	return w.client.MessagingV1().Subscriptions(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
-		ResourceVersion: w.resourceVersion,
-	})
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index c9575d547..694d7d209 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1179,8 +1179,6 @@ knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger
 knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake
 knative.dev/eventing/pkg/client/injection/informers/factory
 knative.dev/eventing/pkg/client/injection/informers/factory/fake
-knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription
-knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger
 knative.dev/eventing/pkg/client/listers/eventing/v1


Test patch:
diff --git a/.github/workflows/kind-e2e.yaml b/.github/workflows/kind-e2e.yaml
index ba6bfc419..86c037c7d 100644
--- a/.github/workflows/kind-e2e.yaml
+++ b/.github/workflows/kind-e2e.yaml
@@ -27,6 +27,8 @@ jobs:
         test-suite:
           - ./test/e2e
           - ./test/e2e_new
+          - ./test/e2e_channel
+          - ./test/e2e_new_channel
 
         # Map between K8s and KinD versions.
         # This is attempting to make it a bit clearer what's being tested.
@@ -38,7 +40,7 @@ jobs:
           - k8s-version: v1.22.2
             kind-version: v0.11.1
             kind-image-sha: sha256:f638a08c1f68fe2a99e724ace6df233a546eaf6713019a0b310130a4f91ebe7f
-          - test-suite: ./test/e2e
+          - test-suite: ./test/e2e_channel
             extra-test-flags: -channels=messaging.knative.dev/v1beta1:KafkaChannel
 
     env:
@@ -123,7 +125,7 @@ jobs:
           export CLUSTER_DOMAIN=${CLUSTER_SUFFIX}
 
           # Run the tests tagged as e2e on the KinD cluster.
-          go test -race -count=1 -timeout=1h -v -short -tags=e2e \
+          go test -race -count=1 -timeout=1h -v -short -tags=e2e,cloudevents \
              ${{ matrix.test-suite }} ${{ matrix.extra-test-flags }}
 
       - name: Collect system diagnostics
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 93dad84b3..8f9bc3b38 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -979,7 +979,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						},
 						reconcilertesting.WithInitBrokerConditions,
 						StatusBrokerDataPlaneAvailable,
-						StatusBrokerConfigParsed,
+						StatusBrokerConfigNotParsed("failed to resolve Spec.Delivery.DeadLetterSink: destination missing Ref and URI, expected at least one"),
 						StatusBrokerTopicReady,
 					),
 				},
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index ffd577dbc..64f9e2831 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,7 +22,6 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
-	eventingrekttesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/network"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -351,9 +350,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -376,11 +372,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -434,9 +425,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -459,11 +447,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -517,9 +500,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -542,11 +522,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -603,12 +578,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
-				eventingrekttesting.NewSubscription(Subscription2Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription2UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -631,22 +600,12 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}, {
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
 								Destination:   "http://" + Subscription2URI,
 								Uid:           Subscription2UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_DiscardReply{},
-								Reference: &contract.Reference{
-									Name:      Subscription2Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription2UUID,
-								},
 							}},
 						},
 					},
@@ -974,9 +933,6 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewSSLSecret(system.Namespace(), "secret-1"),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1007,11 +963,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -1266,9 +1217,8 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
-			Env:                env,
-			ConfigMapLister:    listers.GetConfigMapLister(),
-			SubscriptionLister: listers.GetSubscriptionLister(),
+			Env:             env,
+			ConfigMapLister: listers.GetConfigMapLister(),
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index f3a49c227..cc3ca16ce 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -24,7 +24,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
-	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
diff --git a/control-plane/pkg/reconciler/consumer/controller_test.go b/control-plane/pkg/reconciler/consumer/controller_test.go
new file mode 100644
index 000000000..d2d8e77de
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumer/controller_test.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 consumer
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
+
+	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer/fake"
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+)
+
+func TestNewController(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	dynamicScheme := runtime.NewScheme()
+	_ = fakekubeclientset.AddToScheme(dynamicScheme)
+
+	dynamicclient.With(ctx, dynamicScheme)
+
+	configs := &config.Env{
+		DataPlaneConfigFormat: "json",
+	}
+
+	controller := NewController(ctx, configs)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}
+
+func TestFormatSerDeFromString(t *testing.T) {
+	tt := []struct {
+		format string
+		serde  contract.FormatSerDe
+	}{
+		{
+			format: "protobuf",
+			serde:  contract.FormatSerDe{Format: contract.Protobuf},
+		},
+		{
+			format: "json",
+			serde:  contract.FormatSerDe{Format: contract.Json},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.format, func(t *testing.T) {
+
+			got := formatSerDeFromString(tc.format)
+
+			require.Equal(t, tc.serde, got, "%v %v", tc.serde, got)
+		})
+	}
+
+}
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index d9a36da09..eeedc53ae 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -21,9 +21,11 @@ header "Running tests"
 
 export_logs_continuously
 
-go_test_e2e -timeout=1h ./test/e2e/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E suite failed"
+go_test_e2e -timeout=1h ./test/e2e/... || fail_test "E2E suite failed"
 
-go_test_e2e -tags=deletecm ./test/e2e/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E (deletecm) suite failed"
+go_test_e2e -timeout=1h ./test/e2e_channel/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E suite (KafkaChannel) failed"
+
+go_test_e2e -tags=deletecm ./test/e2e/... || fail_test "E2E (deletecm) suite failed"
 
 if ! ${LOCAL_DEVELOPMENT}; then
   go_test_e2e -tags=sacura -timeout=40m ./test/e2e/... || fail_test "E2E (sacura) suite failed"
diff --git a/test/e2e/conformance/main_test.go b/test/e2e/conformance/main_test.go
index 2b7c1ca53..b5daffbf5 100644
--- a/test/e2e/conformance/main_test.go
+++ b/test/e2e/conformance/main_test.go
@@ -24,24 +24,14 @@ import (
 	"os"
 	"testing"
 
-	"knative.dev/eventing-kafka/test"
-	eventingTest "knative.dev/eventing/test"
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/pkg/system"
 	"knative.dev/pkg/test/zipkin"
 )
 
-var channelTestRunner testlib.ComponentsTestRunner
-
 func TestMain(m *testing.M) {
 
 	os.Exit(func() int {
-		eventingTest.InitializeEventingFlags()
-		channelTestRunner = testlib.ComponentsTestRunner{
-			ComponentFeatureMap: test.ChannelFeatureMap,
-			ComponentsToTest:    eventingTest.EventingFlags.Channels,
-		}
-
 		// Any tests may SetupZipkinTracing, it will only actually be done once. This should be the ONLY
 		// place that cleans it up. If an individual test calls this instead, then it will break other
 		// tests that need the tracing in place.
diff --git a/test/e2e/main_test.go b/test/e2e/main_test.go
index 76196aceb..d8d336e83 100644
--- a/test/e2e/main_test.go
+++ b/test/e2e/main_test.go
@@ -23,21 +23,11 @@ import (
 	"os"
 	"testing"
 
-	"knative.dev/eventing-kafka/test"
-	eventingTest "knative.dev/eventing/test"
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/pkg/system"
 )
 
-var channelTestRunner testlib.ComponentsTestRunner
-
 func TestMain(m *testing.M) {
-
-	eventingTest.InitializeEventingFlags()
-	channelTestRunner = testlib.ComponentsTestRunner{
-		ComponentFeatureMap: test.ChannelFeatureMap,
-		ComponentsToTest:    eventingTest.EventingFlags.Channels,
-	}
 	os.Exit(func() int {
 		defer testlib.ExportLogs(testlib.SystemLogsDir, system.Namespace())
 
diff --git a/test/e2e/channel_event_transformation_test.go b/test/e2e_channel/channel_event_transformation_test.go
similarity index 97%
rename from test/e2e/channel_event_transformation_test.go
rename to test/e2e_channel/channel_event_transformation_test.go
index 07e3d9d8f..be6fb75db 100644
--- a/test/e2e/channel_event_transformation_test.go
+++ b/test/e2e_channel/channel_event_transformation_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_channel
 
 import (
 	"context"
diff --git a/test/e2e/channel_single_event_test.go b/test/e2e_channel/channel_single_event_test.go
similarity index 98%
rename from test/e2e/channel_single_event_test.go
rename to test/e2e_channel/channel_single_event_test.go
index 141752a2a..262c5d2a5 100644
--- a/test/e2e/channel_single_event_test.go
+++ b/test/e2e_channel/channel_single_event_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e
+package e2e_channel
 
 import (
 	"context"
diff --git a/test/e2e/conformance/channel_addressable_resolver_cluster_role_test.go b/test/e2e_channel/conformance/channel_addressable_resolver_cluster_role_test.go
similarity index 100%
rename from test/e2e/conformance/channel_addressable_resolver_cluster_role_test.go
rename to test/e2e_channel/conformance/channel_addressable_resolver_cluster_role_test.go
diff --git a/test/e2e/conformance/channel_channelable_manipulator_cluster_role_test.go b/test/e2e_channel/conformance/channel_channelable_manipulator_cluster_role_test.go
similarity index 100%
rename from test/e2e/conformance/channel_channelable_manipulator_cluster_role_test.go
rename to test/e2e_channel/conformance/channel_channelable_manipulator_cluster_role_test.go
diff --git a/test/e2e/conformance/channel_crd_metadata_test.go b/test/e2e_channel/conformance/channel_crd_metadata_test.go
similarity index 100%
rename from test/e2e/conformance/channel_crd_metadata_test.go
rename to test/e2e_channel/conformance/channel_crd_metadata_test.go
diff --git a/test/e2e/conformance/channel_crd_name_test.go b/test/e2e_channel/conformance/channel_crd_name_test.go
similarity index 100%
rename from test/e2e/conformance/channel_crd_name_test.go
rename to test/e2e_channel/conformance/channel_crd_name_test.go
diff --git a/test/e2e/conformance/channel_spec_test.go b/test/e2e_channel/conformance/channel_spec_test.go
similarity index 100%
rename from test/e2e/conformance/channel_spec_test.go
rename to test/e2e_channel/conformance/channel_spec_test.go
diff --git a/test/e2e/conformance/channel_status_subscriber_test.go b/test/e2e_channel/conformance/channel_status_subscriber_test.go
similarity index 100%
rename from test/e2e/conformance/channel_status_subscriber_test.go
rename to test/e2e_channel/conformance/channel_status_subscriber_test.go
diff --git a/test/e2e/conformance/channel_status_test.go b/test/e2e_channel/conformance/channel_status_test.go
similarity index 100%
rename from test/e2e/conformance/channel_status_test.go
rename to test/e2e_channel/conformance/channel_status_test.go
diff --git a/test/e2e/conformance/channel_tracing_test.go b/test/e2e_channel/conformance/channel_tracing_test.go
similarity index 100%
rename from test/e2e/conformance/channel_tracing_test.go
rename to test/e2e_channel/conformance/channel_tracing_test.go
diff --git a/test/e2e_channel/conformance/main_test.go b/test/e2e_channel/conformance/main_test.go
new file mode 100644
index 000000000..2b7c1ca53
--- /dev/null
+++ b/test/e2e_channel/conformance/main_test.go
@@ -0,0 +1,53 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package conformance
+
+import (
+	"log"
+	"os"
+	"testing"
+
+	"knative.dev/eventing-kafka/test"
+	eventingTest "knative.dev/eventing/test"
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/pkg/system"
+	"knative.dev/pkg/test/zipkin"
+)
+
+var channelTestRunner testlib.ComponentsTestRunner
+
+func TestMain(m *testing.M) {
+
+	os.Exit(func() int {
+		eventingTest.InitializeEventingFlags()
+		channelTestRunner = testlib.ComponentsTestRunner{
+			ComponentFeatureMap: test.ChannelFeatureMap,
+			ComponentsToTest:    eventingTest.EventingFlags.Channels,
+		}
+
+		// Any tests may SetupZipkinTracing, it will only actually be done once. This should be the ONLY
+		// place that cleans it up. If an individual test calls this instead, then it will break other
+		// tests that need the tracing in place.
+		defer zipkin.CleanupZipkinTracingSetup(log.Printf)
+		defer testlib.ExportLogs(testlib.SystemLogsDir, system.Namespace())
+
+		return m.Run()
+	}())
+}
diff --git a/test/e2e_channel/main_test.go b/test/e2e_channel/main_test.go
new file mode 100644
index 000000000..0c67ac791
--- /dev/null
+++ b/test/e2e_channel/main_test.go
@@ -0,0 +1,46 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e_channel
+
+import (
+	"os"
+	"testing"
+
+	"knative.dev/eventing-kafka/test"
+	eventingTest "knative.dev/eventing/test"
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/pkg/system"
+)
+
+var channelTestRunner testlib.ComponentsTestRunner
+
+func TestMain(m *testing.M) {
+
+	eventingTest.InitializeEventingFlags()
+	channelTestRunner = testlib.ComponentsTestRunner{
+		ComponentFeatureMap: test.ChannelFeatureMap,
+		ComponentsToTest:    eventingTest.EventingFlags.Channels,
+	}
+	os.Exit(func() int {
+		defer testlib.ExportLogs(testlib.SystemLogsDir, system.Namespace())
+
+		return m.Run()
+	}())
+}
diff --git a/test/e2e_new/main_test.go b/test/e2e_new/main_test.go
index f8c717918..c31e03102 100644
--- a/test/e2e_new/main_test.go
+++ b/test/e2e_new/main_test.go
@@ -24,8 +24,6 @@ import (
 	"os"
 	"testing"
 
-	"knative.dev/eventing/test/rekt/resources/channel_impl"
-
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
@@ -47,10 +45,6 @@ func init() {
 	// environment.InitFlags registers state and level filter flags.
 	environment.InitFlags(flag.CommandLine)
 
-	// CHANNEL_GROUP_KIND=KafkaChannel.messaging.knative.dev;CHANNEL_VERSION=v1beta1
-	channel_impl.EnvCfg.ChannelGK = "KafkaChannel.messaging.knative.dev"
-	channel_impl.EnvCfg.ChannelV = "v1beta1"
-
 	// Force the broker class to be configured properly
 	broker.EnvCfg.BrokerClass = kafka.BrokerClass
 }
diff --git a/test/e2e_new/kafka_channel_test.go b/test/e2e_new_channel/kafka_channel_test.go
similarity index 99%
rename from test/e2e_new/kafka_channel_test.go
rename to test/e2e_new_channel/kafka_channel_test.go
index 6c9c8f2be..f099c6c73 100644
--- a/test/e2e_new/kafka_channel_test.go
+++ b/test/e2e_new_channel/kafka_channel_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e_new
+package e2e_new_channel
 
 import (
 	"fmt"
diff --git a/test/e2e_new/kafka_channel_upstream_test.go b/test/e2e_new_channel/kafka_channel_upstream_test.go
similarity index 99%
rename from test/e2e_new/kafka_channel_upstream_test.go
rename to test/e2e_new_channel/kafka_channel_upstream_test.go
index 132670950..c7b660760 100644
--- a/test/e2e_new/kafka_channel_upstream_test.go
+++ b/test/e2e_new_channel/kafka_channel_upstream_test.go
@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-package e2e_new
+package e2e_new_channel
 
 import (
 	"testing"
diff --git a/test/e2e_new_channel/main_test.go b/test/e2e_new_channel/main_test.go
new file mode 100644
index 000000000..8e4d6e5de
--- /dev/null
+++ b/test/e2e_new_channel/main_test.go
@@ -0,0 +1,71 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e_new_channel
+
+import (
+	"flag"
+	"os"
+	"testing"
+
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+
+	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
+	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+
+	"knative.dev/pkg/injection"
+	_ "knative.dev/pkg/system/testing"
+
+	"knative.dev/reconciler-test/pkg/environment"
+)
+
+// global is the singleton instance of GlobalEnvironment. It is used to parse
+// the testing config for the test run. The config will specify the cluster
+// config as well as the parsing level and state flags.
+var global environment.GlobalEnvironment
+
+func init() {
+	// environment.InitFlags registers state and level filter flags.
+	environment.InitFlags(flag.CommandLine)
+
+	// CHANNEL_GROUP_KIND=KafkaChannel.messaging.knative.dev;CHANNEL_VERSION=v1beta1
+	channel_impl.EnvCfg.ChannelGK = "KafkaChannel.messaging.knative.dev"
+	channel_impl.EnvCfg.ChannelV = "v1beta1"
+}
+
+// TestMain is the first entry point for `go test`.
+func TestMain(m *testing.M) {
+	// We get a chance to parse flags to include the framework flags for the
+	// framework as well as any additional flags included in the integration.
+	flag.Parse()
+
+	// EnableInjectionOrDie will enable client injection, this is used by the
+	// testing framework for namespace management, and could be leveraged by
+	// features to pull Kubernetes clients or the test environment out of the
+	// context passed in the features.
+	ctx, startInformers := injection.EnableInjectionOrDie(nil, nil) //nolint
+	startInformers()
+
+	// global is used to make instances of Environments, NewGlobalEnvironment
+	// is passing and saving the client injection enabled context for use later.
+	global = environment.NewGlobalEnvironment(ctx)
+
+	// Run the tests.
+	os.Exit(m.Run())
+}
diff --git a/test/reconciler-tests.sh b/test/reconciler-tests.sh
index 72636f544..0f823dfe1 100755
--- a/test/reconciler-tests.sh
+++ b/test/reconciler-tests.sh
@@ -23,6 +23,8 @@ export_logs_continuously
 
 go_test_e2e -timeout=1h ./test/e2e_new/... || fail_test "E2E (new) suite failed"
 
+go_test_e2e -tags=e2e,cloudevents -timeout=1h ./test/e2e_new_channel/... || fail_test "E2E (new - KafkaChannel) suite failed"
+
 go_test_e2e -tags=deletecm ./test/e2e_new/... || fail_test "E2E (new deletecm) suite failed"
 
 if ! ${LOCAL_DEVELOPMENT}; then
diff --git a/test/upgrade/postupgrade.go b/test/upgrade/postupgrade.go
index f555ce771..04cf80a79 100644
--- a/test/upgrade/postupgrade.go
+++ b/test/upgrade/postupgrade.go
@@ -32,7 +32,7 @@ func BrokerPostUpgradeTest() pkgupgrade.Operation {
 
 // SinkPostUpgradeTest tests sink basic operations post upgrade.
 func SinkPostUpgradeTest() pkgupgrade.Operation {
-	return pkgupgrade.NewOperation("SinkPostDowngradeTest", func(c pkgupgrade.Context) {
+	return pkgupgrade.NewOperation("SinkPostUpgradeTest", func(c pkgupgrade.Context) {
 		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
 	})
 }
diff --git a/test/upgrade/preupgrade.go b/test/upgrade/preupgrade.go
index c46b23f41..1df23a95b 100644
--- a/test/upgrade/preupgrade.go
+++ b/test/upgrade/preupgrade.go
@@ -32,7 +32,7 @@ func BrokerPreUpgradeTest() pkgupgrade.Operation {
 
 // SinkPreUpgradeTest tests sink basic operations pre upgrade.
 func SinkPreUpgradeTest() pkgupgrade.Operation {
-	return pkgupgrade.NewOperation("SinkPostDowngradeTest", func(c pkgupgrade.Context) {
+	return pkgupgrade.NewOperation("SinkPreUpgradeTest", func(c pkgupgrade.Context) {
 		e2e.RunTestKafkaSink(c.T, eventing.ModeBinary, nil)
 	})
 }
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index ece583caf..71aafe19a 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20220113-afceb5404"
+    kafka.eventing.knative.dev/release: "v20220131-a14b76602"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crd.yaml b/third_party/eventing-kafka-latest/source-crd.yaml
index c070768e7..905329876 100644
--- a/third_party/eventing-kafka-latest/source-crd.yaml
+++ b/third_party/eventing-kafka-latest/source-crd.yaml
@@ -16,7 +16,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20220113-afceb5404"
+    kafka.eventing.knative.dev/release: "v20220131-a14b76602"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 83f275c51..a46bb3de0 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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -436,10 +436,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     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: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -725,10 +725,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -783,10 +783,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -796,9 +796,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20220122-f07487898"
+        eventing.knative.dev/release: "v20220131-7a44b07f2"
         app.kubernetes.io/component: eventing-autoscaler
-        app.kubernetes.io/version: "20220122-f07487898"
+        app.kubernetes.io/version: "20220131-7a44b07f2"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -816,7 +816,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:d51f4e8a1e19a657fdccde63a465d4a4fd00240faa73d847e13d8c96f327ce87
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:69d2d30440ed84ebc50be23694ca194646fa08411ddbe93487bc003090a52d9c
           resources:
             requests:
               cpu: 100m
@@ -834,7 +834,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:c0bde79565c48e297ebf9c7072b205a7127a5090ca2a3e9285f6456555a45bee
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:72b6206cf15de6cfdb16217a17ec5d8e0184dd3a14c206861ab16ad5d94802ce
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -883,9 +883,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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.
@@ -898,9 +898,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220122-f07487898"
+        eventing.knative.dev/release: "v20220131-7a44b07f2"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20220122-f07487898"
+        app.kubernetes.io/version: "20220131-7a44b07f2"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -914,7 +914,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:17f30435a9f39d3121263186c90927a869437d7d7babe51fe1fbecfcececc9fe
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:744060290509d1172fbaaded191bf1b5b631f835de27662c6bfe62358526ebaa
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -979,9 +979,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1005,9 +1005,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1036,9 +1036,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1049,9 +1049,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220122-f07487898"
+        eventing.knative.dev/release: "v20220131-7a44b07f2"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20220122-f07487898"
+        app.kubernetes.io/version: "20220131-7a44b07f2"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1071,7 +1071,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:ace72d0992e9526b1ea7825eb7647809a4eeb3f6d1f71ff36fc10c92dfd11e0b
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:cff89e978a3ecb1e6a3af700525a832058feccf5ec769ce111da673d9fa9b80f
           resources:
             requests:
               # taken from serving.
@@ -1140,10 +1140,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1175,11 +1175,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1391,10 +1391,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1558,11 +1558,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1830,11 +1830,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1984,9 +1984,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2113,10 +2113,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2368,11 +2368,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2559,10 +2559,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2879,12 +2879,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3071,9 +3071,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3283,9 +3283,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3469,8 +3469,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3483,9 +3483,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3503,9 +3503,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3526,9 +3526,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3553,9 +3553,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3574,9 +3574,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3612,8 +3612,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3639,8 +3639,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3657,8 +3657,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3691,8 +3691,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3705,9 +3705,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3745,9 +3745,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3759,9 +3759,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3773,9 +3773,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3787,9 +3787,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3801,9 +3801,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3816,8 +3816,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3830,8 +3830,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3858,8 +3858,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3994,8 +3994,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4062,8 +4062,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4076,9 +4076,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4124,8 +4124,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4138,9 +4138,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4176,8 +4176,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4263,8 +4263,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4363,8 +4363,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4400,8 +4400,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4438,8 +4438,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4472,8 +4472,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4507,8 +4507,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 # The data is populated at install time.
 
@@ -4532,8 +4532,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
-    app.kubernetes.io/version: "20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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 9c9f22eff..99b7e5a89 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: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     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: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -672,11 +672,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -826,9 +826,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -955,10 +955,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1210,11 +1210,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1401,10 +1401,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1721,12 +1721,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1913,9 +1913,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2125,9 +2125,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20220122-f07487898"
+    eventing.knative.dev/release: "v20220131-7a44b07f2"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220122-f07487898"
+    app.kubernetes.io/version: "20220131-7a44b07f2"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev