Repository: knative-sandbox/eventing-kafka-broker
Pull request #1524 (2021-11-22 11:05:22)

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

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

Details:
```

```

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

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

<details>

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

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


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1524#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **julz** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @julz` in a comment when ready.

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

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

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

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/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/1524?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    #1524      +/-   ##
============================================
+ Coverage     73.40%   82.00%   +8.59%     
  Complexity      586      586              
============================================
  Files           103       66      -37     
  Lines          3993     1900    -2093     
  Branches        165      165              
============================================
- Hits           2931     1558    -1373     
+ Misses          824      256     -568     
+ Partials        238       86     -152     
```

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

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/core/config/resource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvcmVzb3VyY2UuZ28=) | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | |
| [control-plane/pkg/reconciler/kafka/broker\_class.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9icm9rZXJfY2xhc3MuZ28=) | | |
| [...ntrol-plane/pkg/reconciler/broker/broker\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyX2NvbmZpZy5nbw==) | | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | | |
| [control-plane/pkg/reconciler/kafka/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9jbGllbnQuZ28=) | | |
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | | |
| [control-plane/pkg/core/config/utils.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvdXRpbHMuZ28=) | | |
| [...ntrol-plane/pkg/reconciler/source/auth\_provider.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvYXV0aF9wcm92aWRlci5nbw==) | | |
| [control-plane/pkg/security/scram.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2NyYW0uZ28=) | | |
| ... and [27 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524?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/1524?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [809a632...49739f5](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1524?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: @knative-automation: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

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

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

<details>

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

Patch:
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 663002d4a..fea7d0dbd 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -171,10 +171,15 @@ func GetDataPlaneConfigMapData(logger *zap.Logger, dataPlaneConfigMap *corev1.Co
 	}
 	if err != nil {
 
-		logger.Warn("Failed to unmarshal config map", zap.Error(err))
+		// Because of https://github.com/golang/protobuf/issues/1314, we cannot use the error message in the unit tests.
+		// Also, the error here should only happen when there is an implementation issue as the configmap is only
+		// to be touched by the controller.
+		// Thus, let's not have the error message in the returned error. Instead, let's just print it.
+
+		logger.Error("Failed to unmarshal contract", zap.Any("content", dataPlaneDataRaw), zap.Error(err))
 
 		// let the caller decide if it want to continue or fail on an error.
-		return ct, fmt.Errorf("failed to unmarshal contract: '%s' - %w", dataPlaneDataRaw, err)
+		return ct, fmt.Errorf("failed to unmarshal contract: '%s'", dataPlaneDataRaw)
 	}
 
 	return ct, nil


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index fb988ead5..a19edb07e 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -28,7 +28,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
-
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
@@ -84,7 +83,9 @@ kafka:
 const (
 	testProber = "testProber"
 
-	finalizerName = "kafkachannels.messaging.knative.dev"
+	finalizerName                 = "kafkachannels.messaging.knative.dev"
+	TestExpectedDataNumPartitions = "TestExpectedDataNumPartitions"
+	TestExpectedReplicationFactor = "TestExpectedReplicationFactor"
 )
 
 var finalizerUpdatedEvent = Eventf(
@@ -102,10 +103,17 @@ var DefaultEnv = &config.Env{
 	DataPlaneConfigFormat:       base.Json,
 }
 
+// TODO: compare with inmemorychannel_test.go at /Users/aliok/code/github.com/knative/eventing/pkg/reconciler/inmemorychannel/controller/inmemorychannel_test.go
+// TODO: compare with channel_test.go at /Users/aliok/code/github.com/knative/eventing/pkg/reconciler/channel/channel_test.go
 // TODO: tests with and without subscriptions
 // TODO: tests for things from config-kafka
-// TODO: are we setting a InitialOffsetsCommitted status?
-// TODO: test if things from spec is used properly?
+// TODO: are we setting a InitialOffsetsCommitted status? we gotta set it on the subscription. 1) can we do it? 2) does it make sense?
+// TODO: test if things from channel spec is used properly?
+// TODO: test if things from subscription spec is used properly?
+// TODO: rename test cases consistently
+// TODO: test finalize?
+// TODO: fix the order of status updates
+// TODO: add tests where there are no CM updates
 
 func TestReconcileKind(t *testing.T) {
 
@@ -115,6 +123,31 @@ func TestReconcileKind(t *testing.T) {
 	testKey := fmt.Sprintf("%s/%s", ChannelNamespace, ChannelName)
 
 	table := TableTest{
+		{
+			Name: "bad workqueue key",
+			// Make sure Reconcile handles bad keys.
+			Key: "too/many/parts",
+		},
+		{
+			Name: "key not found",
+			// Make sure Reconcile handles good keys that don't exist.
+			Key: "foo/not-found",
+		},
+		{
+			Name: "Channel not found",
+			Key:  testKey,
+		},
+		{
+			Name: "Channel is being deleted",
+			Key:  testKey,
+			Objects: []runtime.Object{
+				NewChannel(
+					WithInitKafkaChannelConditions,
+					WithDeletedTimeStamp),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+		},
 		{
 			Name: "Reconciled normal - no subscription - no auth",
 			Objects: []runtime.Object{
@@ -317,17 +350,739 @@ func TestReconcileKind(t *testing.T) {
 				testProber: probertesting.MockProber(prober.StatusUnknown),
 			},
 		},
-	}
-
-	useTable(t, table, env)
-}
-
-func useTable(t *testing.T, table TableTest, env config.Env) {
-	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
-
-		proberMock := probertesting.MockProber(prober.StatusReady)
-		if p, ok := row.OtherTestData[testProber]; ok {
-			proberMock = p.(prober.Prober)
+		{
+			Name: "Reconciled normal - with single fresh subscriber - no auth",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
+								Destination:   "http://" + Subscription1URI,
+								Uid:           Subscription1UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+							}},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+						WithSubscribers(Subscriber1()),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Reconciled normal - with single unready subscriber - no auth",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithUnreadySubscriber))),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
+								Destination:   "http://" + Subscription1URI,
+								Uid:           Subscription1UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+							}},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+						WithSubscribers(Subscriber1()),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Reconciled normal - with single ready subscriber - no auth",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1())),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
+								Destination:   "http://" + Subscription1URI,
+								Uid:           Subscription1UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+							}},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+						WithSubscribers(Subscriber1()),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Reconciled normal - with two fresh subscribers - no auth",
+			Objects: []runtime.Object{
+				NewChannel(
+					WithSubscribers(Subscriber1(WithFreshSubscriber)),
+					WithSubscribers(Subscriber2(WithFreshSubscriber)),
+				),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
+								Destination:   "http://" + Subscription1URI,
+								Uid:           Subscription1UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+							}, {
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
+								Destination:   "http://" + Subscription2URI,
+								Uid:           Subscription2UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+							}},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+						WithSubscribers(Subscriber1(), Subscriber2()),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Data plane not available - no receiver",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			Key:                     testKey,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantErr:                 true,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusDataPlaneNotAvailable,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf("%s: %s", base.ReasonDataPlaneNotAvailable, base.MessageDataPlaneNotAvailable),
+				),
+			},
+		},
+		{
+			Name: "Data plane not available - no dispatcher",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			Key:                     testKey,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantErr:                 true,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusDataPlaneNotAvailable,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf("%s: %s", base.ReasonDataPlaneNotAvailable, base.MessageDataPlaneNotAvailable),
+				),
+			},
+		},
+		{
+			Name: "config-kafka not resolved",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			Key:                     testKey,
+			WantErr:                 true,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigNotParsed(fmt.Sprintf(`failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), constants.SettingsConfigMapName, constants.SettingsConfigMapName)),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(`failed to get contract configuration: failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), constants.SettingsConfigMapName, constants.SettingsConfigMapName),
+				),
+			},
+		},
+		{
+			Name: "config-kafka not readable",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, map[string]string{
+					"version":        "1.0.0",
+					"eventing-kafka": `123123`,
+				}),
+			},
+			Key:                     testKey,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantErr:                 true,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigNotParsed("ConfigMap's eventing-kafka value could not be converted to an EventingKafkaConfig struct: error unmarshaling JSON: while decoding JSON: json: cannot unmarshal number into Go value of type config.EventingKafkaConfig : 123123"),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"failed to get contract configuration: ConfigMap's eventing-kafka value could not be converted to an EventingKafkaConfig struct: error unmarshaling JSON: while decoding JSON: json: cannot unmarshal number into Go value of type config.EventingKafkaConfig : 123123",
+				),
+			},
+		},
+		{
+			Name: "Channel spec is used properly",
+			Objects: []runtime.Object{
+				NewChannel(
+					WithNumPartitions(3),
+					WithReplicationFactor(4),
+					WithRetentionDuration("1000"),
+				),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			OtherTestData: map[string]interface{}{
+				TestExpectedDataNumPartitions: int32(3),
+				TestExpectedReplicationFactor: int16(4),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithNumPartitions(3),
+						WithReplicationFactor(4),
+						WithRetentionDuration("1000"),
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			// TODO
+			Name: "Reconciled normal - with single fresh subscriber - with auth - PlainText",
+		},
+		{
+			// TODO
+			Name: "Reconciled normal - with single fresh subscriber - with auth - SASL PlainText",
+		},
+		{
+			// TODO
+			Name: "Reconciled normal - with single fresh subscriber - with auth - SSL",
+		},
+		{
+			// TODO
+			Name: "Reconciled normal - with single fresh subscriber - with auth - SASLSSL",
+		},
+		{
+			// TODO
+			Name: "Unknown auth protocol  - with single fresh subscriber - with auth - unknown protocol",
+		},
+		{
+			Name: "Create contract configmap when it does not exist",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Do not create contract configmap when it exists",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Corrupt contract in configmap",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithBinaryData(&env, []byte("corrupt")),
+			},
+			Key:                     testKey,
+			WantErr:                 true,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigParsed,
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusConfigMapNotUpdatedReady(
+							"Failed to get contract data from ConfigMap: knative-eventing/kafka-channel-channels-subscriptions",
+							"failed to unmarshal contract: 'corrupt'",
+						),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"failed to get broker and triggers data from config map knative-eventing/kafka-channel-channels-subscriptions: failed to unmarshal contract: 'corrupt'",
+				),
+			},
+		},
+	}
+
+	useTable(t, table, env)
+}
+
+func useTable(t *testing.T, table TableTest, env config.Env) {
+	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		proberMock := probertesting.MockProber(prober.StatusReady)
+		if p, ok := row.OtherTestData[testProber]; ok {
+			proberMock = p.(prober.Prober)
+		}
+
+		numPartitions := int32(1)
+		if v, ok := row.OtherTestData[TestExpectedDataNumPartitions]; ok {
+			numPartitions = v.(int32)
+		}
+
+		replicationFactor := int16(1)
+		if v, ok := row.OtherTestData[TestExpectedReplicationFactor]; ok {
+			replicationFactor = v.(int16)
 		}
 
 		reconciler := &Reconciler{
@@ -354,8 +1109,8 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName: ChannelTopic(),
 					ExpectedTopicDetail: sarama.TopicDetail{
-						NumPartitions:     1,
-						ReplicationFactor: 1,
+						NumPartitions:     numPartitions,
+						ReplicationFactor: replicationFactor,
 					},
 					T: t,
 				}, nil
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 3690c990e..4b044be76 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -194,7 +194,7 @@ func StatusBrokerConfigParsed(broker *eventing.Broker) {
 
 func StatusBrokerConfigNotParsed(reason string) func(broker *eventing.Broker) {
 	return func(broker *eventing.Broker) {
-		StatusConfigNotParsed(broker, reason)
+		StatusConfigNotParsed(reason)(broker)
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index 1447f43dc..c6de5977c 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -30,6 +30,7 @@ import (
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/network"
@@ -40,6 +41,15 @@ const (
 	ChannelNamespace        = "test-nc"
 	ChannelUUID             = "c1234567-8901-2345-6789-123456789101"
 	ChannelBootstrapServers = "kafka:9092"
+
+	Subscription1Name     = "sub-1"
+	Subscription2Name     = "sub-2"
+	Subscription1UUID     = "2f9b5e8e-deb6-11e8-9f32-f2801f1b9fd1"
+	Subscription2UUID     = "34c5aec8-deb6-11e8-9f32-f2801f1b9fd1"
+	Subscription1URI      = "sub-1-uri"
+	Subscription2URI      = "sub-2-uri"
+	Subscription1ReplyURI = "sub-1-reply-uri"
+	Subscription2ReplyURI = "sub-2-reply-uri"
 )
 
 func ChannelTopic() string {
@@ -62,6 +72,27 @@ func NewChannel(options ...KRShapedOption) runtime.Object {
 	return c
 }
 
+func WithNumPartitions(np int32) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ch := obj.(*messagingv1beta1.KafkaChannel)
+		ch.Spec.NumPartitions = np
+	}
+}
+
+func WithReplicationFactor(rp int16) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ch := obj.(*messagingv1beta1.KafkaChannel)
+		ch.Spec.ReplicationFactor = rp
+	}
+}
+
+func WithRetentionDuration(rd string) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ch := obj.(*messagingv1beta1.KafkaChannel)
+		ch.Spec.RetentionDuration = rd
+	}
+}
+
 func ChannelReceiverPod(namespace string, annotations map[string]string) runtime.Object {
 	return &corev1.Pod{
 		TypeMeta: metav1.TypeMeta{
@@ -147,3 +178,80 @@ func WithInitKafkaChannelConditions(obj duckv1.KRShaped) {
 	channel := obj.(*messagingv1beta1.KafkaChannel)
 	channel.Status.InitializeConditions()
 }
+
+type SubscriberInfo struct {
+	spec   *eventingduckv1.SubscriberSpec
+	status *eventingduckv1.SubscriberStatus
+}
+
+type subscriberInfoOption func(sub *SubscriberInfo)
+
+func WithSubscribers(subscribers ...*SubscriberInfo) func(obj duckv1.KRShaped) {
+	return func(obj duckv1.KRShaped) {
+		channel := obj.(*messagingv1beta1.KafkaChannel)
+		if len(channel.Spec.Subscribers) == 0 {
+			channel.Spec.Subscribers = []eventingduckv1.SubscriberSpec{}
+		}
+
+		if len(channel.Status.Subscribers) == 0 {
+			channel.Status.Subscribers = []eventingduckv1.SubscriberStatus{}
+		}
+
+		for _, s := range subscribers {
+			if s.spec != nil {
+				channel.Spec.Subscribers = append(channel.Spec.Subscribers, *s.spec)
+			}
+			if s.status != nil {
+				channel.Status.Subscribers = append(channel.Status.Subscribers, *s.status)
+			}
+		}
+	}
+}
+
+func Subscriber1(options ...subscriberInfoOption) *SubscriberInfo {
+	s := &SubscriberInfo{
+		spec: &eventingduckv1.SubscriberSpec{
+			UID:           Subscription1UUID,
+			Generation:    1,
+			SubscriberURI: apis.HTTP(Subscription1URI),
+			ReplyURI:      apis.HTTP(Subscription1ReplyURI),
+		},
+		status: &eventingduckv1.SubscriberStatus{
+			UID:                Subscription1UUID,
+			ObservedGeneration: 1,
+			Ready:              "True",
+		},
+	}
+	for _, opt := range options {
+		opt(s)
+	}
+	return s
+}
+
+func Subscriber2(options ...subscriberInfoOption) *SubscriberInfo {
+	s := &SubscriberInfo{
+		spec: &eventingduckv1.SubscriberSpec{
+			UID:           Subscription2UUID,
+			Generation:    1,
+			SubscriberURI: apis.HTTP(Subscription2URI),
+			ReplyURI:      apis.HTTP(Subscription2ReplyURI),
+		},
+		status: &eventingduckv1.SubscriberStatus{
+			UID:                Subscription2UUID,
+			ObservedGeneration: 1,
+			Ready:              "True",
+		},
+	}
+	for _, opt := range options {
+		opt(s)
+	}
+	return s
+}
+
+func WithFreshSubscriber(sub *SubscriberInfo) {
+	sub.status = nil
+}
+
+func WithUnreadySubscriber(sub *SubscriberInfo) {
+	sub.status.Ready = "False"
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 3caea87f2..54960b4c1 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -188,8 +188,10 @@ func StatusConfigParsed(obj duckv1.KRShaped) {
 	obj.GetConditionSet().Manage(obj.GetStatus()).MarkTrue(base.ConditionConfigParsed)
 }
 
-func StatusConfigNotParsed(obj duckv1.KRShaped, reason string) {
-	obj.GetConditionSet().Manage(obj.GetStatus()).MarkFalse(base.ConditionConfigParsed, reason, "")
+func StatusConfigNotParsed(reason string) func(obj duckv1.KRShaped) {
+	return func(obj duckv1.KRShaped) {
+		obj.GetConditionSet().Manage(obj.GetStatus()).MarkFalse(base.ConditionConfigParsed, reason, "")
+	}
 }
 
 func StatusConfigMapUpdatedReady(env *config.Env) func(obj duckv1.KRShaped) {
@@ -202,6 +204,16 @@ func StatusConfigMapUpdatedReady(env *config.Env) func(obj duckv1.KRShaped) {
 	}
 }
 
+func StatusConfigMapNotUpdatedReady(reason, message string) func(obj duckv1.KRShaped) {
+	return func(obj duckv1.KRShaped) {
+		obj.GetConditionSet().Manage(obj.GetStatus()).MarkFalse(
+			base.ConditionConfigMapUpdated,
+			reason,
+			message,
+		)
+	}
+}
+
 func StatusTopicReadyWithName(topic string) func(obj duckv1.KRShaped) {
 	return func(obj duckv1.KRShaped) {
 		obj.GetConditionSet().Manage(obj.GetStatus()).MarkTrueWithReason(
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 75deac027..3677b9711 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: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -50,8 +50,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -67,8 +67,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -84,8 +84,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -101,8 +101,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -118,8 +118,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -151,8 +151,8 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: 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: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -193,8 +193,8 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -202,8 +202,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -220,8 +220,8 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -237,8 +237,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -254,8 +254,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -287,8 +287,8 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 data:
   channelTemplateSpec: |
@@ -316,8 +316,8 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Configures the default for any Broker that does not specify a spec.config or Broker class.
@@ -350,8 +350,8 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
@@ -385,10 +385,10 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 data:
   _example: |
@@ -432,10 +432,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -476,7 +476,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -525,8 +525,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "96896b00"
@@ -587,10 +587,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -641,10 +641,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -718,10 +718,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -776,10 +776,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/high-availability: "true"
     app.kubernetes.io/name: eventing-controller
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -789,9 +789,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20211116-d5201e7b4"
+        eventing.knative.dev/release: "v20211121-4e372e524"
         app.kubernetes.io/name: eventing-autoscaler
-        app.kubernetes.io/version: "20211116-d5201e7b4"
+        app.kubernetes.io/version: "20211121-4e372e524"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -809,7 +809,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:d5247ac61afcb44fb1cbcaba3fbb57843dd568205c2efd550081973e2c958379
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:efffbfd5c3fece2d64f6c37c09564f788ffa7e2659a45eac190a5272b32e9d5b
           resources:
             requests:
               cpu: 100m
@@ -827,7 +827,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:4728b752e57d98bcf5f8501e7cb2a81ac7df14159c53c2b004b5b184caab22d4
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:3644e39fbaf7b38b19274861e79f80fbf853e5bd3cc8432924e43d095963a2c4
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -876,9 +876,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     app.kubernetes.io/name: pingsource-mt-adapter
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
@@ -891,9 +891,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211116-d5201e7b4"
+        eventing.knative.dev/release: "v20211121-4e372e524"
         app.kubernetes.io/name: pingsource-mt-adapter
-        app.kubernetes.io/version: "20211116-d5201e7b4"
+        app.kubernetes.io/version: "20211121-4e372e524"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       affinity:
@@ -907,7 +907,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:d5b391248aa16a613c054e931f695442f7d632f089f62e132195061e5f7d6282
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:9c3becba624549e39de4542fc3700681299388b2f3c1a4431dc1fdab88596bee
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -972,9 +972,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   scaleTargetRef:
@@ -998,9 +998,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   minAvailable: 80%
@@ -1029,9 +1029,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -1042,9 +1042,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211116-d5201e7b4"
+        eventing.knative.dev/release: "v20211121-4e372e524"
         app.kubernetes.io/name: eventing-webhook
-        app.kubernetes.io/version: "20211116-d5201e7b4"
+        app.kubernetes.io/version: "20211121-4e372e524"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1064,7 +1064,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:59027542302047bc06251cb60c17906d483b0356e1e3eb7adc49cc87a1e537da
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:55e5b7b9cd4ab0d440a5cffbbfe312b0b3158be1d16754d0871ecf127fec866b
           resources:
             requests:
               # taken from serving.
@@ -1133,10 +1133,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     role: eventing-webhook
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1168,11 +1168,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1384,10 +1384,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1551,11 +1551,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1823,11 +1823,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1977,9 +1977,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2106,10 +2106,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2361,11 +2361,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2552,10 +2552,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2871,12 +2871,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3063,9 +3063,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3275,9 +3275,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3461,8 +3461,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3475,9 +3475,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3495,9 +3495,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3518,9 +3518,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3545,9 +3545,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3566,9 +3566,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3604,8 +3604,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3631,8 +3631,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3649,8 +3649,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3683,8 +3683,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3697,9 +3697,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3737,9 +3737,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3751,9 +3751,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3765,9 +3765,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3779,9 +3779,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3793,9 +3793,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3808,8 +3808,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3822,8 +3822,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3850,8 +3850,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3986,8 +3986,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4054,8 +4054,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4068,9 +4068,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4116,8 +4116,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4130,9 +4130,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4168,8 +4168,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4255,8 +4255,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4355,8 +4355,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4392,8 +4392,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4430,8 +4430,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4464,8 +4464,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4499,8 +4499,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 # The data is populated at install time.
 
@@ -4524,8 +4524,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: 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 f0ed1f7c3..97b02f389 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: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: 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: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -400,11 +400,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -672,11 +672,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: 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: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -955,10 +955,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1210,11 +1210,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: 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: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1720,12 +1720,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1912,9 +1912,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2124,9 +2124,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211116-d5201e7b4"
+    eventing.knative.dev/release: "v20211121-4e372e524"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211116-d5201e7b4"
+    app.kubernetes.io/version: "20211121-4e372e524"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev