Repository: knative-sandbox/eventing-kafka-broker
Pull request #1396 (2021-11-01 11:05:10)

[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/1396):

>/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/1396#" 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 **benmoss** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @benmoss` 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":["benmoss"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1396](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c57b076) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/b36793cead0d4c165ad207e6460d2626c207307b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b36793c) will **decrease** coverage by `6.54%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/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/1396?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    #1396      +/-   ##
============================================
- Coverage     76.53%   69.99%   -6.55%     
============================================
  Files            99       34      -65     
  Lines          3597     1673    -1924     
  Branches        165        0     -165     
============================================
- Hits           2753     1171    -1582     
+ Misses          634      379     -255     
+ Partials        210      123      -87     
```

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

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...fka/broker/dispatcher/impl/filter/ExactFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvRXhhY3RGaWx0ZXIuamF2YQ==) | | |
| [...ve/eventing/kafka/broker/dispatcher/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9NYWluLmphdmE=) | | |
| [...ka/broker/dispatcher/impl/filter/PrefixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvUHJlZml4RmlsdGVyLmphdmE=) | | |
| [...oker/dispatcher/main/ConsumerDeployerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lckRlcGxveWVyVmVydGljbGUuamF2YQ==) | | |
| [...ka/broker/dispatcher/impl/NoopResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9Ob29wUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [...ka/broker/dispatcher/impl/filter/SuffixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvU3VmZml4RmlsdGVyLmphdmE=) | | |
| [...kafka/broker/dispatcher/impl/filter/AnyFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQW55RmlsdGVyLmphdmE=) | | |
| [...receiver/impl/handler/MethodNotAllowedHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvTWV0aG9kTm90QWxsb3dlZEhhbmRsZXIuamF2YQ==) | | |
| [...nciler/impl/ResourcesReconcilerMessageHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJNZXNzYWdlSGFuZGxlci5qYXZh) | | |
| [...tive/eventing/kafka/broker/core/utils/BaseEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQmFzZUVudi5qYXZh) | | |
| ... and [55 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396/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/1396?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/1396?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [b36793c...c57b076](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1396?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-integration-tests | c57b076fecf6f80f4dce62b02fbe8f7560d078a7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1396/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1455128784112979968) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1396). [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 -->
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-integration-tests | c57b076fecf6f80f4dce62b02fbe8f7560d078a7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1396/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1455421504807243776) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1396). [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 -->
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/1396#" title="Author self-approved">knative-automation</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1396#pullrequestreview-794952445" title="Approved">matzew</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]

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: New changes are detected. LGTM label has been removed.
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-integration-tests | df52084a1fb24d2958410affc6fb38963c1bdfa8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1396/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1455491452116144128) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1396). [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/core/config/egress.go b/control-plane/pkg/core/config/egress.go
index e18425e05..56a186875 100644
--- a/control-plane/pkg/core/config/egress.go
+++ b/control-plane/pkg/core/config/egress.go
@@ -47,11 +47,11 @@ const (
 )
 
 // AddOrUpdateEgressConfig adds or updates the given egress to the given contract at the specified indexes.
-func AddOrUpdateEgressConfig(ct *contract.Contract, brokerIndex int, egress *contract.Egress, egressIndex int) int {
+func AddOrUpdateEgressConfig(ct *contract.Contract, resourceIndex int, egress *contract.Egress, egressIndex int) int {
 
 	if egressIndex != NoEgress {
-		prev := ct.Resources[brokerIndex].Egresses[egressIndex]
-		ct.Resources[brokerIndex].Egresses[egressIndex] = egress
+		prev := ct.Resources[resourceIndex].Egresses[egressIndex]
+		ct.Resources[resourceIndex].Egresses[egressIndex] = egress
 
 		if proto.Equal(prev, egress) {
 			return EgressUnchanged
@@ -59,14 +59,32 @@ func AddOrUpdateEgressConfig(ct *contract.Contract, brokerIndex int, egress *con
 		return EgressChanged
 	}
 
-	ct.Resources[brokerIndex].Egresses = append(
-		ct.Resources[brokerIndex].Egresses,
+	ct.Resources[resourceIndex].Egresses = append(
+		ct.Resources[resourceIndex].Egresses,
 		egress,
 	)
 
 	return EgressChanged
 }
 
+// AddOrUpdateEgressConfigForResource adds or updates the given egress to the given contract at the specified indexes.
+func AddOrUpdateEgressConfigForResource(resource *contract.Resource, egress *contract.Egress, egressIndex int) int {
+
+	if egressIndex != NoEgress {
+		prev := resource.Egresses[egressIndex]
+		resource.Egresses[egressIndex] = egress
+
+		if proto.Equal(prev, egress) {
+			return EgressUnchanged
+		}
+		return EgressChanged
+	}
+
+	resource.Egresses = append(resource.Egresses, egress)
+
+	return EgressChanged
+}
+
 // KeyTypeFromString returns the contract.KeyType associated to a given string.
 func KeyTypeFromString(s string) contract.KeyType {
 	switch s {
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 29503c905..ac3b55333 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -65,9 +65,9 @@ type Reconciler struct {
 	bootstrapServersLock         sync.RWMutex
 	ConfigMapLister              corelisters.ConfigMapLister
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
 	Configs *Configs
 }
@@ -127,18 +127,18 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 
 	topicName := kafka.Topic(TopicPrefix, broker)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin config: %w", err))
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdmin, logger, topicName, topicConfig)
+	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topic, err)
 	}
@@ -299,22 +299,22 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	// get security option for Sarama with secret info in it
 	securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
 		// topic undeleted e.g. when we lose connection
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
 		// topic undeleted e.g. when we lose connection
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.DeleteTopic(kafkaClusterAdmin, kafka.Topic(TopicPrefix, broker))
+	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, kafka.Topic(TopicPrefix, broker))
 	if err != nil {
 		return err
 	}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 2b38c2d4a..8b23ebc89 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -71,7 +71,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
 		},
-		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		KafkaDefaultTopicDetails: sarama.TopicDetail{
 			NumPartitions:     DefaultNumPartitions,
 			ReplicationFactor: DefaultReplicationFactor,
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 1799a49f0..ab4ac8725 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -23,25 +23,28 @@ import (
 	"strings"
 
 	"github.com/Shopify/sarama"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 	"knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 	commonsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/system"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
@@ -58,9 +61,13 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// mock the function used during the reconciliation loop.
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
+
+	// NewKafkaClient creates new sarama Client. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClient kafka.NewClientFunc
 
 	ConfigMapLister corelisters.ConfigMapLister
 
@@ -135,35 +142,49 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 
 	topicName := topic(TopicPrefix, channel)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(saramaSecurityOption)
+	kafkaClusterAdminSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin sarama config: %w", err))
+	}
+
+	// Manually commit the offsets in KafkaChannel controller.
+	// That's because we want to make sure we initialize the offsets within the controller
+	// before dispatcher actually starts consuming messages.
+	kafkaClientSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption, kafka.DisableOffsetAutoCommitConfigOption)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin sarama config: %w", err))
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, kafkaClusterAdminSaramaConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
+
+	kafkaClient, err := r.NewKafkaClient(topicConfig.BootstrapServers, kafkaClientSaramaConfig)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting sarama config: %w", err))
+	}
+	defer kafkaClient.Close()
 
 	// create the topic
-	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdmin, logger, topicName, topicConfig)
+	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topic, err)
 	}
 	logger.Debug("Topic created", zap.Any("topic", topic))
 	statusConditionManager.TopicReady(topic)
 
-	// Get contract config map to write into it.
+	// Get data plane config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
-		return statusConditionManager.FailedToGetConfigMap(err)
+		return statusConditionManager.FailedToGetConfig(err)
 	}
 	logger.Debug("Got contract config map")
 
-	// Get contract data
+	// Get data plane config data.
 	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
-	if err != nil && ct == nil {
+	if err != nil || ct == nil {
 		return statusConditionManager.FailedToGetDataFromConfigMap(err)
 	}
 	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
@@ -175,7 +196,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	}
 	coreconfig.SetDeadLetterSinkURIFromEgressConfig(&channel.Status.DeliveryStatus, channelResource.EgressConfig)
 
-	// Update contract data with the new contract configuration
+	// we still update the contract configMap even though there's an error.
+	// however, we record this error to make the reconciler try again.
+	subscriptionError := r.reconcileSubscribers(ctx, kafkaClient, kafkaClusterAdminClient, channel, channelResource)
+
+	// Update contract data with the new contract configuration (add/update channel resource)
 	channelIndex := coreconfig.FindResource(ct, channel.UID)
 	changed := coreconfig.AddOrUpdateResourceConfig(ct, channelResource, channelIndex, logger)
 	logger.Debug("Change detector", zap.Int("changed", changed))
@@ -229,6 +254,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		logger.Debug("Updated dispatcher pod annotation")
 	}
 
+	if subscriptionError != nil {
+		logger.Error("Error reconciling subscriptions. Going to try again.", zap.Error(subscriptionError))
+		return subscriptionError
+	}
+
 	return statusConditionManager.Reconciled()
 }
 
@@ -327,22 +357,22 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 	// get security option for Sarama with secret info in it
 	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(secret)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(saramaSecurityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
 		// topic undeleted e.g. when we lose connection
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
 		// topic undeleted e.g. when we lose connection
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.DeleteTopic(kafkaClusterAdmin, topic(TopicPrefix, channel))
+	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, topic(TopicPrefix, channel))
 	if err != nil {
 		return err
 	}
@@ -352,6 +382,91 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 	return nil
 }
 
+func (r *Reconciler) reconcileSubscribers(ctx context.Context, kafkaClient sarama.Client, kafkaClusterAdmin sarama.ClusterAdmin, channel *messagingv1beta1.KafkaChannel, channelContractResource *contract.Resource) error {
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, channel)
+
+	channel.Status.Subscribers = make([]v1.SubscriberStatus, 0)
+	var globalErr error
+	for i := range channel.Spec.Subscribers {
+		s := &channel.Spec.Subscribers[i]
+		logger = logger.With(zap.Any("subscription", s))
+		err := r.reconcileSubscriber(ctx, kafkaClient, kafkaClusterAdmin, channel, s, channelContractResource)
+		if err != nil {
+			logger.Error("error reconciling subscription. marking subscription as not ready", zap.Error(err))
+			channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+				UID:                s.UID,
+				ObservedGeneration: s.Generation,
+				Ready:              corev1.ConditionFalse,
+				Message:            fmt.Sprint("Subscription not ready", err),
+			})
+			globalErr = multierr.Append(globalErr, err)
+		} else {
+			logger.Debug("marking subscription as ready")
+			channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+				UID:                s.UID,
+				ObservedGeneration: s.Generation,
+				Ready:              corev1.ConditionTrue,
+			})
+		}
+	}
+	return globalErr
+}
+
+func (r *Reconciler) reconcileSubscriber(ctx context.Context, kafkaClient sarama.Client, kafkaClusterAdmin sarama.ClusterAdmin, channel *messagingv1beta1.KafkaChannel, subscriberSpec *v1.SubscriberSpec, channelContractResource *contract.Resource) error {
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, channel)
+
+	logger.Debug("Reconciling initial offset for subscription", zap.Any("subscription", subscriberSpec), zap.Any("channel", channel))
+	err := r.reconcileInitialOffset(ctx, channel, subscriberSpec, kafkaClient, kafkaClusterAdmin)
+	if err != nil {
+		return fmt.Errorf("initial offset cannot be committed: %v", err)
+	}
+	logger.Debug("Reconciled initial offset for subscription. ", zap.Any("subscription", subscriberSpec))
+
+	subscriberIndex := coreconfig.FindEgress(channelContractResource.Egresses, subscriberSpec.UID)
+	subscriberConfig, err := r.getSubscriberConfig(ctx, channel, subscriberSpec)
+	if err != nil {
+		return fmt.Errorf("failed to resolve subscriber config: %v", zap.Error(err))
+	}
+
+	coreconfig.AddOrUpdateEgressConfigForResource(channelContractResource, subscriberConfig, subscriberIndex)
+	return nil
+}
+
+func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec, kafkaClient sarama.Client, kafkaClusterAdmin sarama.ClusterAdmin) error {
+	subscriptionStatus := findSubscriptionStatus(channel, sub.UID)
+	if subscriptionStatus != nil && subscriptionStatus.Ready == corev1.ConditionTrue {
+		// subscription is ready, the offsets must have been initialized already
+		return nil
+	}
+
+	topicName := topic(TopicPrefix, channel)
+	groupID := consumerGroup(channel, sub)
+	_, err := offset.InitOffsets(ctx, kafkaClient, kafkaClusterAdmin, []string{topicName}, groupID)
+	return err
+}
+
+func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
+	egress := &contract.Egress{
+		Destination:   subscriber.SubscriberURI.String(),
+		ConsumerGroup: consumerGroup(channel, subscriber),
+		Uid:           string(subscriber.UID),
+	}
+
+	subscriptionEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, subscriber.Delivery, r.Configs.DefaultBackoffDelayMs)
+	if err != nil {
+		return nil, err
+	}
+	channelEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, channel.Spec.Delivery, r.Configs.DefaultBackoffDelayMs)
+	if err != nil {
+		return nil, err
+	}
+
+	// Merge Channel and Subscription egress configuration prioritizing the Subscription configuration.
+	egress.EgressConfig = coreconfig.MergeEgressConfig(subscriptionEgressConfig, channelEgressConfig)
+
+	return egress, nil
+}
+
 func (r *Reconciler) channelConfigMap() (*corev1.ConfigMap, error) {
 	// TODO: do we want to support namespaced channels? they're not supported at the moment.
 
@@ -422,8 +537,22 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
-// Topic returns a topic name given a topic prefix and a generic object.
+// topic returns a topic name given a topic prefix and a generic object.
 // This function uses a different format than the kafkatopic.Topic function
 func topic(prefix string, obj metav1.Object) string {
 	return fmt.Sprintf("%s.%s.%s", prefix, obj.GetNamespace(), obj.GetName())
 }
+
+// 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))
+}
+
+func findSubscriptionStatus(kc *messagingv1beta1.KafkaChannel, subUID types.UID) *v1.SubscriberStatus {
+	for _, subStatus := range kc.Status.Subscribers {
+		if subStatus.UID == subUID {
+			return &subStatus
+		}
+	}
+	return nil
+}
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index dd7cb44c2..56eb80218 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -60,9 +60,10 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
-		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
-		Configs:              configs,
-		ConfigMapLister:      configmapInformer.Lister(),
+		NewKafkaClient:             sarama.NewClient,
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		Configs:                    configs,
+		ConfigMapLister:            configmapInformer.Lister(),
 	}
 
 	logger := logging.FromContext(ctx)
diff --git a/control-plane/pkg/reconciler/kafka/client.go b/control-plane/pkg/reconciler/kafka/client.go
index fdf598b29..b129275b5 100644
--- a/control-plane/pkg/reconciler/kafka/client.go
+++ b/control-plane/pkg/reconciler/kafka/client.go
@@ -36,15 +36,23 @@ func NoOpConfigOption(*sarama.Config) error {
 	return nil
 }
 
-// NewClusterAdminFunc creates new sarama.ClusterAdmin.
-type NewClusterAdminFunc func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)
+func DisableOffsetAutoCommitConfigOption(config *sarama.Config) error {
+	config.Consumer.Offsets.AutoCommit.Enable = false
+	return nil
+}
+
+// NewClusterAdminClientFunc creates new sarama.ClusterAdmin.
+type NewClusterAdminClientFunc func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)
+
+// NewClientFunc creates new sarama.Client.
+type NewClientFunc func(addrs []string, config *sarama.Config) (sarama.Client, error)
 
-// GetClusterAdminSaramaConfig returns Kafka Admin configurations that the ConfigOptions are applied to
-func GetClusterAdminSaramaConfig(configOption ConfigOption) (*sarama.Config, error) {
+// GetSaramaConfig returns Kafka Client configuration with the given options applied.
+func GetSaramaConfig(configOptions ...ConfigOption) (*sarama.Config, error) {
 	config := sarama.NewConfig()
 	config.Version = sarama.MaxVersion
 
-	err := configOption(config)
+	err := Options(config, configOptions...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index c47e41965..aa4784957 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -58,9 +58,9 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			SystemNamespace:             configs.SystemNamespace,
 			ReceiverLabel:               base.SinkReceiverLabel,
 		},
-		ConfigMapLister:      configmapInformer.Lister(),
-		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
-		Configs:              configs,
+		ConfigMapLister:            configmapInformer.Lister(),
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		Configs:                    configs,
 	}
 
 	_, err := reconciler.GetOrCreateDataPlaneConfigMap(ctx)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index bee978514..26e0744e0 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -48,9 +48,9 @@ type Reconciler struct {
 
 	ConfigMapLister corelisters.ConfigMapLister
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
 	Configs *config.Env
 }
@@ -100,16 +100,16 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		return fmt.Errorf("failed to track secret: %w", err)
 	}
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(ks.Spec.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(ks.Spec.BootstrapServers, saramaConfig)
 	if err != nil {
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
 	if ks.Spec.NumPartitions != nil && ks.Spec.ReplicationFactor != nil {
 
@@ -117,7 +117,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 
 		topicConfig := topicConfigFromSinkSpec(&ks.Spec)
 
-		topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdmin, logger, ks.Spec.Topic, topicConfig)
+		topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, ks.Spec.Topic, topicConfig)
 		if err != nil {
 			return statusConditionManager.FailedToCreateTopic(topic, err)
 		}
@@ -126,7 +126,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		// If the topic is externally managed, we need to make sure that the topic exists and it's valid.
 		ks.GetStatus().Annotations[base.TopicOwnerAnnotation] = ExternalTopicOwner
 
-		isPresentAndValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdmin, ks.Spec.Topic)
+		isPresentAndValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdminClient, ks.Spec.Topic)
 		if err != nil {
 			return statusConditionManager.TopicsNotPresentOrInvalidErr([]string{ks.Spec.Topic}, err)
 		}
@@ -286,22 +286,22 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 		// get security option for Sarama with secret info in it
 		securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
 
-		saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+		saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 		if err != nil {
 			// even in error case, we return `normal`, since we are fine with leaving the
 			// topic undeleted e.g. when we lose connection
 			return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 		}
 
-		kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(ks.Spec.BootstrapServers, saramaConfig)
+		kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(ks.Spec.BootstrapServers, saramaConfig)
 		if err != nil {
 			// even in error case, we return `normal`, since we are fine with leaving the
 			// topic undeleted e.g. when we lose connection
 			return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 		}
-		defer kafkaClusterAdmin.Close()
+		defer kafkaClusterAdminClient.Close()
 
-		topic, err := kafka.DeleteTopic(kafkaClusterAdmin, ks.Spec.Topic)
+		topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, ks.Spec.Topic)
 		if err != nil {
 			return err
 		}
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
index 123c6000c..875e81831 100644
--- a/control-plane/pkg/reconciler/source/controller.go
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -50,8 +50,8 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			SystemNamespace:             configs.SystemNamespace,
 			DispatcherLabel:             base.SourceDispatcherLabel,
 		},
-		Env:                  configs,
-		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
+		Env:                        configs,
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 	}
 
 	impl := kafkasource.NewImpl(ctx, r)
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 125cecb34..133c5d3b7 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -57,9 +57,9 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
@@ -99,18 +99,18 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		return fmt.Errorf("failed to track secrets: %w", err)
 	}
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(ks.Spec.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(ks.Spec.BootstrapServers, saramaConfig)
 	if err != nil {
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	isValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdmin, ks.Spec.Topics...)
+	isValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdminClient, ks.Spec.Topics...)
 	if err != nil {
 		return statusConditionManager.TopicsNotPresentOrInvalidErr(ks.Spec.Topics, err)
 	}
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index befcbe197..1d6290ec6 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -20,8 +20,8 @@ Lists of 153 third-party dependencies.
      (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.1.1-android - https://github.com/google/guava/guava)
      (The Apache Software License, Version 2.0) Guava ListenableFuture only (com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava - https://github.com/google/guava/listenablefuture)
      (The Apache Software License, Version 2.0) J2ObjC Annotations (com.google.j2objc:j2objc-annotations:1.3 - https://github.com/google/j2objc/)
-     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.0 - https://developers.google.com/protocol-buffers/protobuf-java/)
-     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.0 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
+     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.1 - https://developers.google.com/protocol-buffers/protobuf-java/)
+     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.1 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
      (Apache 2.0) OkHttp Logging Interceptor (com.squareup.okhttp3:logging-interceptor:3.12.12 - https://github.com/square/okhttp/logging-interceptor)
      (Apache 2.0) MockWebServer (com.squareup.okhttp3:mockwebserver:3.12.6 - https://github.com/square/okhttp/mockwebserver)
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 6c8c43358..fd8234347 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -46,7 +46,7 @@
     <micrometer.version>1.7.5</micrometer.version>
     <opentelemetry.version>1.7.0</opentelemetry.version>
     <jackson.version>2.13.0</jackson.version>
-    <protobuf.version>3.19.0</protobuf.version>
+    <protobuf.version>3.19.1</protobuf.version>
     <slf4j.version>1.7.32</slf4j.version>
     <ch.qos.logback.version>1.2.6</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>6.6</net.logstash.logback.encoder.version>
diff --git a/go.mod b/go.mod
index 6fe198b8e..acfcc71b2 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,7 @@ require (
 	github.com/rickb777/date v1.14.1
 	github.com/stretchr/testify v1.7.0
 	github.com/xdg-go/scram v1.0.2
+	go.uber.org/multierr v1.6.0
 	go.uber.org/zap v1.19.1
 	google.golang.org/protobuf v1.27.1
 	k8s.io/api v0.21.4
@@ -21,9 +22,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca
-	knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c
-	knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e
-	knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40
-	knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c
+	knative.dev/eventing v0.27.0
+	knative.dev/eventing-kafka v0.27.0
+	knative.dev/hack v0.0.0-20211101195839-11d193bf617b
+	knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc
+	knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13
 )
diff --git a/go.sum b/go.sum
index 058e124f9..f3f86d054 100644
--- a/go.sum
+++ b/go.sum
@@ -24,7 +24,6 @@ cloud.google.com/go v0.87.0/go.mod h1:TpDYlFy7vuLzZMMZ+B6iRiELaY7z/gJPaqbMx6mlWc
 cloud.google.com/go v0.90.0/go.mod h1:kRX0mNRHe0e2rC6oNakvwQqzyDmg57xJ+SZU1eT2aDQ=
 cloud.google.com/go v0.93.3/go.mod h1:8utlLll2EF5XMAV15woO4lSbWQlk8rer9aLOfLh7+YI=
 cloud.google.com/go v0.94.1/go.mod h1:qAlAugsXlC+JWO+Bke5vCtc9ONxjQT3drlTTnAplMW4=
-cloud.google.com/go v0.96.0/go.mod h1:GF7l59pYBVlXQIBLx3a761cZ41F9bBH3JUlihCt2Udc=
 cloud.google.com/go v0.97.0 h1:3DXvAyifywvq64LfkKaMOmkWPS1CikIQdMe2lY9vxU8=
 cloud.google.com/go v0.97.0/go.mod h1:GF7l59pYBVlXQIBLx3a761cZ41F9bBH3JUlihCt2Udc=
 cloud.google.com/go/bigquery v1.0.1/go.mod h1:i/xbL2UlR5RvWAURpBYZTtm/cXjCha9lbfbpx4poX+o=
@@ -45,7 +44,6 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
-cloud.google.com/go/storage v1.16.1/go.mod h1:LaNorbty3ehnU3rEjXSNV/NRgQA0O8Y+uh6bPe5UOk4=
 cloud.google.com/go/storage v1.18.2 h1:5NQw6tOn3eMm0oE8vTkfjau18kjL79FlMjy/CHTpmoY=
 cloud.google.com/go/storage v1.18.2/go.mod h1:AiIj7BWXyhO5gGVmYJ+S8tbkCx3yb0IMjua8Aw4naVM=
 contrib.go.opencensus.io/exporter/ocagent v0.7.1-0.20200907061046-05415f1de66d h1:LblfooH1lKOpp1hIhukktmSAxFkqMPFk9KR6iZ0MJNI=
@@ -613,7 +611,6 @@ github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y8
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
 github.com/prometheus/common v0.26.0/go.mod h1:M7rCNAaPfAosfx8veZJCuw84e35h3Cfd9VFqTh1DIvc=
 github.com/prometheus/common v0.28.0/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
-github.com/prometheus/common v0.30.0/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
 github.com/prometheus/common v0.31.1 h1:d18hG4PkHnNAKNMOmFuXFaiY8Us0nird/2m60uS1AMs=
 github.com/prometheus/common v0.31.1/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
@@ -740,7 +737,6 @@ go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/automaxprocs v1.3.0/go.mod h1:9CWT6lKIep8U41DDaPiH6eFscnTyjfTANNQNx6LrIcA=
 go.uber.org/automaxprocs v1.4.0 h1:CpDZl6aOlLhReez+8S3eEotD7Jx0Os++lemPlMULQP0=
 go.uber.org/automaxprocs v1.4.0/go.mod h1:/mTEdr7LvHhs0v7mjdxDreTz1OG5zdZGqgOnhWiR/+Q=
-go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723 h1:sHOAIxRGBp443oHZIPB+HsUGaksVCXVQENPxwTfQdH4=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
@@ -750,7 +746,6 @@ go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9i
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.15.0/go.mod h1:Mb2vm2krFEG5DV0W9qcHBYFtp/Wku1cvYaqPsS/WYfc=
-go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
 go.uber.org/zap v1.19.1 h1:ue41HOKd1vGURxrmeKIgELGb3jPW9DMUDGtsinblHwI=
 go.uber.org/zap v1.19.1/go.mod h1:j3DNczoxDZroyBnOT1L/Q79cfUMGZxlv/9dzN7SM1rI=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -865,8 +860,6 @@ golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210825183410-e898025ed96a/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f h1:OfiFi4JbukWwe3lzw+xunroH1mnC1e2Gy5cxNJApiSY=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -1018,7 +1011,6 @@ golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191113191852-77e3bb0ad9e7/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191115202509-3a792d9c32b2/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -1158,7 +1150,6 @@ google.golang.org/genproto v0.0.0-20210303154014-9728d6b83eeb/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20210310155132-4ce2db91004e/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210319143718-93e7006c17a6/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210402141018-6c239bbf2bb1/go.mod h1:9lPAdzaEmUacj36I+k7YKbEc5CXzPIeORRgDAUOu28A=
-google.golang.org/genproto v0.0.0-20210416161957-9910b6c460de/go.mod h1:P3QM42oQyzQSnHPnZ/vqoCdDmzH28fzWByN9asMeM8A=
 google.golang.org/genproto v0.0.0-20210513213006-bf773b8c8384/go.mod h1:P3QM42oQyzQSnHPnZ/vqoCdDmzH28fzWByN9asMeM8A=
 google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
 google.golang.org/genproto v0.0.0-20210604141403-392c879c8b08/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
@@ -1170,7 +1161,6 @@ google.golang.org/genproto v0.0.0-20210728212813-7823e685a01f/go.mod h1:ob2IJxKr
 google.golang.org/genproto v0.0.0-20210805201207-89edb61ffb67/go.mod h1:ob2IJxKrgPT52GcgX759i1sleT07tiKowYBGbczaW48=
 google.golang.org/genproto v0.0.0-20210813162853-db860fec028c/go.mod h1:cFeNkxwySK631ADgubI+/XFU/xp8FD5KIVV4rj8UC5w=
 google.golang.org/genproto v0.0.0-20210821163610-241b8fcbd6c8/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
-google.golang.org/genproto v0.0.0-20210825212027-de86158e7fda/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210828152312-66f60bf46e71/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210831024726-fe130286e0e2/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210903162649-d08c68adba83/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
@@ -1288,7 +1278,6 @@ k8s.io/component-base v0.21.4/go.mod h1:ZKG0eHVX+tUDcaoIGpU3Vtk4TIjMddN9uhEWDmW6
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/gengo v0.0.0-20201203183100-97869a43a9d9/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/gengo v0.0.0-20201214224949-b6c5ce23f027/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
-k8s.io/gengo v0.0.0-20210203185629-de9496dff47b/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/gengo v0.0.0-20210915205010-39e73c8a59cd h1:WEhFhsVrgII9WzLqf1VTUUtwPHm+vxbRq8Ib6y9Jt4g=
 k8s.io/gengo v0.0.0-20210915205010-39e73c8a59cd/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/klog v0.2.0/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
@@ -1304,29 +1293,18 @@ k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iL
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20211019131736-762838ee9058/go.mod h1:9mkxOcLsgMaA+LTojeXLpVoN1PW5jogk53pBXzrOQU4=
-knative.dev/eventing v0.26.1-0.20211027202426-729e39fd4218/go.mod h1:27s/uaaUrAUKO/Wy7kR4R6mbj5zugQWCyEkod4dVw0I=
-knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca h1:wELJavauMmbGQQ257fbAkIcZKl+AzH9fwZtD87556iA=
-knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca/go.mod h1:27s/uaaUrAUKO/Wy7kR4R6mbj5zugQWCyEkod4dVw0I=
-knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c h1:MIf1yj66j5c+eM6eHGYMjY15UosCRDsaXIVYoioGf5E=
-knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c/go.mod h1:Vz6hgwmnzvwOtOMsbqrqY8UoteyiwbRtlLV/POAdNLg=
-knative.dev/hack v0.0.0-20210806075220-815cd312d65c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211018110626-47ac3b032e60/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211026141922-a71c865b5f66/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211027200727-f1228dd5e3e6/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e h1:0Hw2xdWYbcs2JRJnOLzAVh7APOtgro7gSno0228mnDg=
-knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20211026141922-a71c865b5f66/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20210914164111-4857ab6939e3/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
-knative.dev/pkg v0.0.0-20211018141937-a34efd6b409d/go.mod h1:/d1+Wklc8blR39M4oEBu+GeLav4vsiiJwq61tJay7mc=
-knative.dev/pkg v0.0.0-20211027105800-3b33e02e5b9c/go.mod h1:xtF+ujTgEybZxA+e7NwKgHR8xMjYzHQUBxyaOUEFxwY=
-knative.dev/pkg v0.0.0-20211027171921-f7b70f5ce303/go.mod h1:xtF+ujTgEybZxA+e7NwKgHR8xMjYzHQUBxyaOUEFxwY=
-knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40 h1:eOOOqcj+IHIp5te9GGZsrq6vCXUv2y5JItRT3nXHs20=
-knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40/go.mod h1:HyEqMTLzT2hGCisGaRMGzEE8b4Ym+7mhmVx5ygiX+fY=
-knative.dev/reconciler-test v0.0.0-20211018075026-816dbd1c57f7/go.mod h1:6yDmb26SINSmgw6wVy9qQwgRMewiW8ddkkwGLR0ZvOY=
-knative.dev/reconciler-test v0.0.0-20211019133535-04ea7ddac7e3/go.mod h1:ltMDlEIFo5/9ES92BhIWQkdZaHRnGGJvXcYlvPxZj5s=
-knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c h1:SQ5ag46cbR0ZCJ49zc4F/+hGs0JdudsLaLpt6vSSX+E=
-knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c/go.mod h1:PIwtS0at/rtnsNCxcZUAZUmWKKasTUYJ/w1Mk94bOdA=
+knative.dev/control-protocol v0.0.0-20211101215039-a32569595694/go.mod h1:HlIoGfErGnSVKN6NI3iG5gXxCo9EcdV8sCDGW2IObJo=
+knative.dev/eventing v0.27.0 h1:P9rqKvsCeFb8sEhfK32WFTG7awdOAa7XO6p3umSq/wU=
+knative.dev/eventing v0.27.0/go.mod h1:4ppWQEQ/2B66/YFENDmV1Gjxs4meLpz6UTUgLkkINt4=
+knative.dev/eventing-kafka v0.27.0 h1:mfbQbK+g9PiBHL9FoLTNckQRvY4EzMfmR2z25zTKJeU=
+knative.dev/eventing-kafka v0.27.0/go.mod h1:bt9HdIsI9RMgiWfcsrDT2vH8ADGY+ehN4a6Bj4QH5Jo=
+knative.dev/hack v0.0.0-20211101195839-11d193bf617b h1:DaW1iliZlBAwq/I8gTqVu8UnfGxyb5yR7CDsJi5jyWk=
+knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack/schema v0.0.0-20211101195839-11d193bf617b/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
+knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc h1:ldjbTpoMXUTgzw0IJsAFLdyA6/6QYRvz8IGPZOknEDg=
+knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
+knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13 h1:PgMCfHECOrYopm3P9s2Xq0Hl/MeBI0RoZ/dBH0a3X2g=
+knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
diff --git a/vendor/knative.dev/eventing-kafka/pkg/common/kafka/offset/offsets.go b/vendor/knative.dev/eventing-kafka/pkg/common/kafka/offset/offsets.go
new file mode 100644
index 000000000..5caab1698
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/common/kafka/offset/offsets.go
@@ -0,0 +1,139 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package offset
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/zap"
+	"knative.dev/pkg/logging"
+
+	knsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+)
+
+// We want to make sure that ALL consumer group offsets are set before marking
+// the resource as ready, to avoid "losing" events in case the consumer group session
+// is closed before at least one message is consumed from ALL partitions.
+// Without InitOffsets, an event sent to a partition with an uninitialized offset
+// will not be forwarded when the session is closed (or a rebalancing is in progress).
+func InitOffsets(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
+	offsetManager, err := sarama.NewOffsetManagerFromClient(consumerGroup, kafkaClient)
+	if err != nil {
+		return -1, err
+	}
+
+	totalPartitions, topicPartitions, err := retrieveAllPartitions(topics, kafkaClient)
+	if err != nil {
+		return -1, err
+	}
+
+	// Fetch topic offsets
+	topicOffsets, err := knsarama.GetOffsets(kafkaClient, topicPartitions, sarama.OffsetNewest)
+	if err != nil {
+		return -1, fmt.Errorf("failed to get the topic offsets: %w", err)
+	}
+
+	// Look for uninitialized offset (-1)
+	offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(consumerGroup, topicPartitions)
+	if err != nil {
+		return -1, err
+	}
+
+	dirty := false
+	for topic, partitions := range offsets.Blocks {
+		for partitionID, block := range partitions {
+			if block.Offset == -1 { // not initialized?
+				partitionsOffsets, ok := topicOffsets[topic]
+				if !ok {
+					// topic may have been deleted. ignore.
+					continue
+				}
+
+				offset, ok := partitionsOffsets[partitionID]
+				if !ok {
+					// partition may have been deleted. ignore.
+					continue
+				}
+
+				logging.FromContext(ctx).Infow("initializing offset", zap.String("topic", topic), zap.Int32("partition", partitionID), zap.Int64("offset", offset))
+
+				pm, err := offsetManager.ManagePartition(topic, partitionID)
+				if err != nil {
+					return -1, fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partitionID, err)
+				}
+
+				pm.MarkOffset(offset, "")
+				dirty = true
+			}
+		}
+	}
+
+	if dirty {
+		offsetManager.Commit()
+		logging.FromContext(ctx).Infow("consumer group offsets committed", zap.String("consumergroup", consumerGroup))
+	}
+
+	// At this stage the resource is considered Ready
+	return int32(totalPartitions), nil
+
+}
+
+func CheckIfAllOffsetsInitialized(kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (bool, error) {
+	_, topicPartitions, err := retrieveAllPartitions(topics, kafkaClient)
+	if err != nil {
+		return false, err
+	}
+
+	// Look for uninitialized offset (-1)
+	offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(consumerGroup, topicPartitions)
+	if err != nil {
+		return false, err
+	}
+
+	for _, partitions := range offsets.Blocks {
+		for _, block := range partitions {
+			if block.Offset == -1 { // not initialized?
+				return false, nil
+			}
+		}
+	}
+
+	return true, nil
+}
+
+func retrieveAllPartitions(topics []string, kafkaClient sarama.Client) (int, map[string][]int32, error) {
+	totalPartitions := 0
+
+	// Retrieve all partitions
+	topicPartitions := make(map[string][]int32)
+	for _, topic := range topics {
+		partitions, err := kafkaClient.Partitions(topic)
+		totalPartitions += len(partitions)
+		if err != nil {
+			return -1, nil, fmt.Errorf("failed to get partitions for topic %s: %w", topic, err)
+		}
+
+		// return a copy of the partitions array in the map
+		// Sarama is caching this array and we don't want nobody to mess with it
+		clone := make([]int32, len(partitions))
+		copy(clone, partitions)
+		topicPartitions[topic] = clone
+	}
+	return totalPartitions, topicPartitions, nil
+}
diff --git a/vendor/knative.dev/hack/release.sh b/vendor/knative.dev/hack/release.sh
index 80b035fb4..0f7c47324 100644
--- a/vendor/knative.dev/hack/release.sh
+++ b/vendor/knative.dev/hack/release.sh
@@ -131,10 +131,18 @@ function master_version() {
   echo "${tokens[0]}.${tokens[1]}"
 }
 
+# Return the minor version of a release.
+# For example, "v0.2.1" returns "2"
+# Parameters: $1 - release version label.
+function minor_version() {
+  local tokens=(${1//\./ })
+  echo "${tokens[1]}"
+}
+
 # Return the release build number of a release.
 # For example, "v0.2.1" returns "1".
 # Parameters: $1 - release version label.
-function release_build_number() {
+function patch_version() {
   local tokens=(${1//\./ })
   echo "${tokens[2]}"
 }
@@ -171,7 +179,10 @@ function prepare_auto_release() {
   PUBLISH_RELEASE=1
 
   git fetch --all || abort "error fetching branches/tags from remote"
-  local tags="$(git tag | cut -d 'v' -f2 | cut -d '.' -f1-2 | sort -V | uniq)"
+  # Support two different formats for tags
+  # - knative-v1.0.0
+  # - v1.0.0
+  local tags="$(git tag | cut -d '-' -f2 | cut -d 'v' -f2 | cut -d '.' -f1-2 | sort -V | uniq)"
   local branches="$( { (git branch -r | grep upstream/release-) ; (git branch | grep release-); } | cut -d '-' -f2 | sort -V | uniq)"
 
   echo "Versions released (from tags): [" "${tags}" "]"
@@ -210,7 +221,10 @@ function prepare_dot_release() {
   git fetch --all || abort "error fetching branches/tags from remote"
   # List latest release
   local releases # don't combine with the line below, or $? will be 0
-  releases="$(hub_tool release)"
+  # Support tags in two formats
+  # - knative-v1.0.0
+  # - v1.0.0
+  releases="$(hub_tool release | cut -d '-' -f2)"
   echo "Current releases are: ${releases}"
   [[ $? -eq 0 ]] || abort "cannot list releases"
   # If --release-branch passed, restrict to that release
@@ -234,7 +248,9 @@ function prepare_dot_release() {
   [[ -n "${major_minor_version}" ]] || abort "cannot get release major/minor version"
   # Ensure there are new commits in the branch, otherwise we don't create a new release
   setup_branch
-  local last_release_commit="$(git rev-list -n 1 "${last_version}")"
+  # Use the original tag (ie. potentially with a knative- prefix) when determining the last version commit sha
+  local github_tag="$(hub_tool release | grep "${last_version}")"
+  local last_release_commit="$(git rev-list -n 1 "${github_tag}")"
   local release_branch_commit="$(git rev-list -n 1 upstream/"${RELEASE_BRANCH}")"
   [[ -n "${last_release_commit}" ]] || abort "cannot get last release commit"
   [[ -n "${release_branch_commit}" ]] || abort "cannot get release branch last commit"
@@ -246,13 +262,13 @@ function prepare_dot_release() {
     exit 0
   fi
   # Create new release version number
-  local last_build="$(release_build_number "${last_version}")"
+  local last_build="$(patch_version "${last_version}")"
   RELEASE_VERSION="${major_minor_version}.$(( last_build + 1 ))"
   echo "Will create release ${RELEASE_VERSION} at commit ${release_branch_commit}"
   # If --release-notes not used, copy from the latest release
   if [[ -z "${RELEASE_NOTES}" ]]; then
     RELEASE_NOTES="$(mktemp)"
-    hub_tool release show -f "%b" "${last_version}" > "${RELEASE_NOTES}"
+    hub_tool release show -f "%b" "${github_tag}" > "${RELEASE_NOTES}"
     echo "Release notes from ${last_version} copied to ${RELEASE_NOTES}"
   fi
 }
@@ -595,6 +611,8 @@ function publish_to_github() {
   local description="$(mktemp)"
   local attachments_dir="$(mktemp -d)"
   local commitish=""
+  local github_tag="knative-${TAG}"
+
   # Copy files to a separate dir
   for artifact in $@; do
     cp ${artifact} "${attachments_dir}"/
@@ -604,8 +622,22 @@ function publish_to_github() {
   if [[ -n "${RELEASE_NOTES}" ]]; then
     cat "${RELEASE_NOTES}" >> "${description}"
   fi
-  git tag -a "${TAG}" -m "${title}"
-  git_push tag "${TAG}"
+  git tag -a "${github_tag}" -m "${title}"
+  git_push tag "${github_tag}"
+
+  # Include a tag for the go module version
+  #
+  # v1.0.0 = v0.27.0
+  # v1.0.1 = v0.27.1
+  # v1.1.1 = v0.28.1
+  #
+  # See: https://github.com/knative/hack/pull/97
+  if [[ "$TAG" == "v1"* ]]; then
+    local release_minor=$(minor_version $TAG)
+    local go_module_version="v0.$(( release_minor + 27 )).$(patch_version $TAG)"
+    git tag -a "${go_module_version}" -m "${title}"
+    git_push tag "${go_module_version}"
+  fi
 
   [[ -n "${RELEASE_BRANCH}" ]] && commitish="--commitish=${RELEASE_BRANCH}"
   for i in {2..0}; do
@@ -613,7 +645,7 @@ function publish_to_github() {
         ${attachments[@]} \
         --file="${description}" \
         "${commitish}" \
-        "${TAG}" && return 0
+        "${github_tag}" && return 0
     if [[ "${i}" -gt 0 ]]; then
       echo "Error publishing the release, retrying in 15s..."
       sleep 15
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 15777f163..90f9c9146 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -359,6 +359,7 @@ go.uber.org/automaxprocs/internal/cgroups
 go.uber.org/automaxprocs/internal/runtime
 go.uber.org/automaxprocs/maxprocs
 # go.uber.org/multierr v1.6.0
+## explicit
 go.uber.org/multierr
 # go.uber.org/zap v1.19.1
 ## explicit
@@ -1030,7 +1031,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca
+# knative.dev/eventing v0.27.0
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1139,7 +1140,7 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c
+# knative.dev/eventing-kafka v0.27.0
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1187,11 +1188,12 @@ knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1
 knative.dev/eventing-kafka/pkg/common/client
 knative.dev/eventing-kafka/pkg/common/config
 knative.dev/eventing-kafka/pkg/common/constants
+knative.dev/eventing-kafka/pkg/common/kafka/offset
 knative.dev/eventing-kafka/pkg/common/kafka/sarama
-# knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e
+# knative.dev/hack v0.0.0-20211101195839-11d193bf617b
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40
+# knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1284,7 +1286,7 @@ knative.dev/pkg/webhook/json
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c
+# knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/control-plane/pkg/core/config/egress_test.go b/control-plane/pkg/core/config/egress_test.go
index 29f8c404b..8b893db1c 100644
--- a/control-plane/pkg/core/config/egress_test.go
+++ b/control-plane/pkg/core/config/egress_test.go
@@ -221,6 +221,96 @@ func TestAddOrUpdateEgressConfig(t *testing.T) {
 	}
 }
 
+func TestAddOrUpdateEgressConfigForResource(t *testing.T) {
+	tests := []struct {
+		name        string
+		givenRes    *contract.Resource
+		egress      *contract.Egress
+		egressIndex int
+		changed     int
+		wantRes     *contract.Resource
+	}{
+		{
+			name: "Egress found - changed",
+			givenRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "xyz",
+					},
+				},
+			},
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: 0,
+			wantRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+		},
+		{
+			name: "Egress found - unchanged",
+			givenRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: 0,
+			wantRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+			changed: EgressUnchanged,
+		},
+		{
+			name: "Egress not found",
+			givenRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: NoEgress,
+			wantRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+					{
+						Uid: "abc",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := AddOrUpdateEgressConfigForResource(tt.givenRes, tt.egress, tt.egressIndex); got != tt.changed {
+				t.Errorf("AddOrUpdateEgressConfig() = %v, want %v", got, tt.changed)
+			}
+
+			if diff := cmp.Diff(tt.wantRes, tt.givenRes, protocmp.Transform()); diff != "" {
+				t.Errorf("(-want, +got) %s", diff)
+			}
+		})
+	}
+}
+
 func TestKeyTypeFromString(t *testing.T) {
 	tests := []struct {
 		s    string
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index aed4d72f6..b913ddc82 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -1887,7 +1887,7 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 			KafkaDefaultTopicDetails:     defaultTopicDetail,
 			KafkaDefaultTopicDetailsLock: sync.RWMutex{},
 			ConfigMapLister:              listers.GetConfigMapLister(),
-			NewKafkaClusterAdmin: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
 					ExpectedTopicDetail: expectedTopicDetail,
diff --git a/control-plane/pkg/reconciler/kafka/topic_test.go b/control-plane/pkg/reconciler/kafka/topic_test.go
index 0fb3a4a38..7694d7cfe 100644
--- a/control-plane/pkg/reconciler/kafka/topic_test.go
+++ b/control-plane/pkg/reconciler/kafka/topic_test.go
@@ -312,7 +312,7 @@ func TestCreateTopicTopicAlreadyExists(t *testing.T) {
 	assert.Nil(t, err, "expected nil error on topic already exists")
 }
 
-func TestNewClusterAdminFuncIsTopicPresent(t *testing.T) {
+func TestNewClusterAdminClientFuncIsTopicPresent(t *testing.T) {
 	tests := []struct {
 		name         string
 		clusterAdmin sarama.ClusterAdmin
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 6629ad98b..787d352df 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -1091,7 +1091,7 @@ func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 				ReceiverLabel:               base.SinkReceiverLabel,
 			},
 			ConfigMapLister: listers.GetConfigMapLister(),
-			NewKafkaClusterAdmin: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:                      expectedTopicName,
 					ExpectedTopicDetail:                    expectedTopicDetail,
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 1c3fc7b3d..2aa8128a6 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -457,7 +457,7 @@ func useTable(t *testing.T, table TableTest, configs broker.Configs) {
 				DispatcherLabel:             base.SourceDispatcherLabel,
 			},
 			Env: &configs.Env,
-			NewKafkaClusterAdmin: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:                      "",
 					ExpectedTopicDetail:                    sarama.TopicDetail{},
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 23ac2842d..3d2b7ecad 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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -472,7 +472,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -521,8 +521,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "96896b00"
@@ -583,10 +583,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -637,10 +637,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -714,10 +714,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "4002b4c2"
@@ -777,10 +777,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/high-availability: "true"
     app.kubernetes.io/name: eventing-controller
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -790,9 +790,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20211028-2eb07cbdd"
+        eventing.knative.dev/release: "v20211101-4de0da062"
         app.kubernetes.io/name: eventing-autoscaler
-        app.kubernetes.io/version: "20211028-2eb07cbdd"
+        app.kubernetes.io/version: "20211101-4de0da062"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -810,7 +810,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:570deacf9cb8c9768c800fc7853abd56f8a3aa478d804223224ddd462c1662c2
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:cceca901eb6ef4ec77af13bddef31acb8514ae6e4485b438d794cc6c6b7a0189
           resources:
             requests:
               cpu: 100m
@@ -828,7 +828,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:328aefcff1d37d31e3d113291fc38be0e3266d9e4158e62956bab8982bafaf12
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:aee503b7e1f77d7c57ce400ae3bac78af547fe962f1293d9f0678d0d9a8ca19c
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -872,9 +872,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     app.kubernetes.io/name: pingsource-mt-adapter
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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.
@@ -887,9 +887,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211028-2eb07cbdd"
+        eventing.knative.dev/release: "v20211101-4de0da062"
         app.kubernetes.io/name: pingsource-mt-adapter
-        app.kubernetes.io/version: "20211028-2eb07cbdd"
+        app.kubernetes.io/version: "20211101-4de0da062"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       affinity:
@@ -903,7 +903,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:486742ebf0fb7cf26cc50be336351c583d546f9a52195bcd297d5d6f5b74d71d
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:fe92717308d82e7df5e2baf318583aef981f59f8d40b2c2038e81762898f3f42
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -961,9 +961,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   scaleTargetRef:
@@ -987,9 +987,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   minAvailable: 80%
@@ -1018,9 +1018,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -1031,9 +1031,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211028-2eb07cbdd"
+        eventing.knative.dev/release: "v20211101-4de0da062"
         app.kubernetes.io/name: eventing-webhook
-        app.kubernetes.io/version: "20211028-2eb07cbdd"
+        app.kubernetes.io/version: "20211101-4de0da062"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1053,7 +1053,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:301b6c69691ec28d75f228f2b88d9bea8b875dd6081c6a82fdbe8869c13c4300
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:2e0af11711334c51e13f6c04f32952e2c60ffc66946f6f07f5d705121dbe3dda
           resources:
             requests:
               # taken from serving.
@@ -1117,10 +1117,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     role: eventing-webhook
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1152,11 +1152,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1368,10 +1368,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1535,11 +1535,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1807,11 +1807,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1961,9 +1961,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2090,10 +2090,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2345,11 +2345,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2536,10 +2536,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2857,12 +2857,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3049,9 +3049,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3261,9 +3261,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3446,8 +3446,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3460,9 +3460,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3480,9 +3480,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: 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: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3530,9 +3530,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3551,9 +3551,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3589,8 +3589,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3616,8 +3616,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3634,8 +3634,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3668,8 +3668,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3682,9 +3682,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3722,9 +3722,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3736,9 +3736,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3750,9 +3750,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3764,9 +3764,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3778,9 +3778,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3793,8 +3793,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3807,8 +3807,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3835,8 +3835,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3971,8 +3971,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4039,8 +4039,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4053,9 +4053,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4101,8 +4101,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4115,9 +4115,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4153,8 +4153,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4240,8 +4240,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4340,8 +4340,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4377,8 +4377,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4415,8 +4415,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4449,8 +4449,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4484,8 +4484,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 # The data is populated at install time.
 
@@ -4509,8 +4509,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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 3f6f01485..5965ebc3d 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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     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: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1722,12 +1722,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1914,9 +1914,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2126,9 +2126,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    eventing.knative.dev/release: "v20211101-4de0da062"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211101-4de0da062"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev