Repository: knative-sandbox/eventing-kafka-broker
Pull request #1295 (2021-10-08 14:09:00)

Add KafkaSource reconciler
Part of #312

## Proposed Changes

- Add KafkaSource reconcile

## TODO

- reconciler tests (see https://github.com/knative-sandbox/eventing-kafka-broker/pull/1298)

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/1295#" title="Author self-approved">pierDipi</a>*

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

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

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

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 97686c8184b367e8efc9d75d9869f51373e6769e | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1295/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1446477743104462848) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Patch:
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index 5392d06f0..b2c244be9 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -40,6 +40,7 @@ const (
 var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 
 	eventingv1alpha1.SchemeGroupVersion.WithKind("KafkaSink"): &eventingv1alpha1.KafkaSink{},
+	// sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"): &sourcesv1beta1.KafkaSource{},
 }
 
 var callbacks = map[schema.GroupVersionKind]validation.Callback{}
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index ec4f43372..63f78dc3c 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -39,7 +39,7 @@ const (
 	ConditionConfigParsed       apis.ConditionType = "ConfigParsed"
 )
 
-var ConditionSet = apis.NewLivingConditionSet(
+var IngressConditionSet = apis.NewLivingConditionSet(
 	ConditionAddressable,
 	ConditionDataPlaneAvailable,
 	ConditionTopicReady,
@@ -47,6 +47,13 @@ var ConditionSet = apis.NewLivingConditionSet(
 	ConditionConfigParsed,
 )
 
+var EgressConditionSet = apis.NewLivingConditionSet(
+	ConditionDataPlaneAvailable,
+	ConditionTopicReady,
+	ConditionConfigMapUpdated,
+	ConditionConfigParsed,
+)
+
 const (
 	TopicOwnerAnnotation = "eventing.knative.dev/topic.owner"
 
@@ -170,14 +177,16 @@ func (manager *StatusConditionManager) TopicReady(topic string) {
 
 func (manager *StatusConditionManager) Reconciled() reconciler.Event {
 
-	object := manager.Object
+	if manager.SetAddress != nil {
+		object := manager.Object
 
-	manager.SetAddress(&apis.URL{
-		Scheme: "http",
-		Host:   network.GetServiceHostname(manager.Configs.IngressName, manager.Configs.SystemNamespace),
-		Path:   fmt.Sprintf("/%s/%s", object.GetNamespace(), object.GetName()),
-	})
-	object.GetConditionSet().Manage(object.GetStatus()).MarkTrue(ConditionAddressable)
+		manager.SetAddress(&apis.URL{
+			Scheme: "http",
+			Host:   network.GetServiceHostname(manager.Configs.IngressName, manager.Configs.SystemNamespace),
+			Path:   fmt.Sprintf("/%s/%s", object.GetNamespace(), object.GetName()),
+		})
+		object.GetConditionSet().Manage(object.GetStatus()).MarkTrue(ConditionAddressable)
+	}
 
 	return nil
 }
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 744ce4043..0ae56b6ae 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -55,7 +55,7 @@ const (
 
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *Configs) *controller.Impl {
 
-	eventing.RegisterAlternateBrokerConditionSet(base.ConditionSet)
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
 	configmapInformer := configmapinformer.Get(ctx)
 
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index d8e41b1dc..650f93c52 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -44,7 +44,7 @@ import (
 
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *config.Env) *controller.Impl {
 
-	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.ConditionSet)
+	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
 
 	configmapInformer := configmapinformer.Get(ctx)
 
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index aa97b8145..19478b0c5 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -41,7 +41,7 @@ import (
 
 func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env) *controller.Impl {
 
-	eventing.RegisterConditionSet(base.ConditionSet)
+	eventing.RegisterConditionSet(base.IngressConditionSet)
 
 	logger := logging.FromContext(ctx)
 
diff --git a/control-plane/pkg/reconciler/source/auth_provider.go b/control-plane/pkg/reconciler/source/auth_provider.go
new file mode 100644
index 000000000..a677d63c6
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/auth_provider.go
@@ -0,0 +1,50 @@
+/*
+ * 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 source
+
+import (
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+)
+
+type SecretLocator struct {
+	*sources.KafkaSource
+}
+
+func (ks *SecretLocator) SecretName() (string, bool) {
+	if !hasAuthConfig(ks.KafkaSource) {
+		return "", false
+	}
+	return getName(ks.KafkaSource), hasAuthConfig(ks.KafkaSource)
+}
+
+func (ks *SecretLocator) SecretNamespace() (string, bool) {
+	return ks.Namespace, hasAuthConfig(ks.KafkaSource)
+}
+
+func getName(ks *sources.KafkaSource) string {
+	return ""
+}
+
+func hasAuthConfig(ks *sources.KafkaSource) bool {
+	// TODO handle secrets
+	//return (ks.Spec.KafkaAuthSpec.Net.TLS.Enable || ks.Spec.KafkaAuthSpec.Net.SASL.Enable) &&
+	//	(ks.Spec.KafkaAuthSpec.Net.TLS.Cert.SecretKeyRef != nil ||
+	//		ks.Spec.KafkaAuthSpec.Net.SASL.User.SecretKeyRef != nil ||
+	//		ks.Spec.KafkaAuthSpec.Net.SASL.Password.SecretKeyRef != nil ||
+	//		ks.Spec.KafkaAuthSpec.Net.SASL.Type.SecretKeyRef != nil)
+	return false
+}
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
new file mode 100644
index 000000000..8f06ce8b8
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -0,0 +1,67 @@
+/*
+ * 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 source
+
+import (
+	"context"
+
+	"github.com/Shopify/sarama"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/resolver"
+
+	kafkainformer "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+)
+
+func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env) *controller.Impl {
+
+	sources.RegisterAlternateKafkaConditionSet(base.EgressConditionSet)
+
+	r := &Reconciler{
+		Reconciler: &base.Reconciler{
+			KubeClient:                  kubeclient.Get(ctx),
+			PodLister:                   podinformer.Get(ctx).Lister(),
+			SecretLister:                secretinformer.Get(ctx).Lister(),
+			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
+			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
+			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			SystemNamespace:             configs.SystemNamespace,
+			DispatcherLabel:             base.BrokerDispatcherLabel,
+			ReceiverLabel:               base.BrokerReceiverLabel,
+		},
+		Env:          configs,
+		ClusterAdmin: sarama.NewClusterAdmin,
+	}
+
+	impl := kafkasource.NewImpl(ctx, r)
+
+	r.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
+
+	kafkaInformer := kafkainformer.Get(ctx)
+
+	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+
+	return impl
+}
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
new file mode 100644
index 000000000..49d15af90
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -0,0 +1,193 @@
+/*
+ * 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 source
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	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"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
+)
+
+type Reconciler struct {
+	*base.Reconciler
+	*config.Env
+
+	Resolver *resolver.URIResolver
+
+	// ClusterAdmin 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.
+	ClusterAdmin kafka.NewClusterAdminFunc
+}
+
+func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
+
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, ks)
+
+	statusConditionManager := base.StatusConditionManager{
+		Object:   ks,
+		Configs:  r.Env,
+		Recorder: controller.GetEventRecorder(ctx),
+	}
+
+	secret, err := security.Secret(ctx, &SecretLocator{KafkaSource: ks}, r.SecretProviderFunc())
+	if err != nil {
+		return fmt.Errorf("failed to get secret: %w", err)
+	}
+	if secret != nil {
+		logger.Debug("Secret reference",
+			zap.String("apiVersion", secret.APIVersion),
+			zap.String("name", secret.Name),
+			zap.String("namespace", secret.Namespace),
+			zap.String("kind", secret.Kind),
+		)
+	}
+
+	// get security option for Sarama with secret info in it
+	securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+
+	if err := r.TrackSecret(secret, ks); err != nil {
+		return fmt.Errorf("failed to track secret: %w", err)
+	}
+
+	for _, t := range ks.Spec.Topics {
+		_, err := r.ClusterAdmin.IsTopicPresentAndValid(t, ks.Spec.BootstrapServers, securityOption)
+		if err != nil {
+			return fmt.Errorf("failed to verify topic validity: %w", err)
+		}
+	}
+
+	// Get contract config map.
+	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return statusConditionManager.FailedToGetConfigMap(err)
+	}
+
+	logger.Debug("Got contract config map")
+
+	// Get contract data.
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	if err != nil && ct == nil {
+		return statusConditionManager.FailedToGetDataFromConfigMap(err)
+	}
+
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
+
+	// Get resource configuration.
+	resource, err := r.getResource(ctx, ks, secret)
+	if err != nil {
+		return statusConditionManager.FailedToGetConfig(err)
+	}
+
+	brokerIndex := coreconfig.FindResource(ct, ks.GetUID())
+	// Update contract data with the new contract configuration
+	changed := coreconfig.AddOrUpdateResourceConfig(ct, resource, brokerIndex, logger)
+
+	logger.Debug("Change detector", zap.Int("changed", changed))
+
+	if changed == coreconfig.ResourceChanged {
+		// Resource changed, increment contract generation.
+		coreconfig.IncrementContractGeneration(ct)
+
+		// Update the configuration map with the new contract data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			logger.Error("failed to update data plane config map", zap.Error(
+				statusConditionManager.FailedToUpdateConfigMap(err),
+			))
+			return err
+		}
+		logger.Debug("Contract config map updated")
+	}
+	statusConditionManager.ConfigMapUpdated()
+
+	// We update dispatcher pods annotation regardless of our contract changed or not due to the fact
+	// that in a previous reconciliation we might have failed to update one of our data plane pod annotation, so we want
+	// to anyway update remaining annotations with the contract generation that was saved in the CM.
+
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+		// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Broker
+		// ready. So, log out the error and move on to the next step.
+		logger.Warn(
+			"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+			zap.Error(err),
+		)
+
+		statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
+	} else {
+		logger.Debug("Updated dispatcher pod annotation")
+	}
+
+	return statusConditionManager.Reconciled()
+}
+
+func (r *Reconciler) getResource(ctx context.Context, ks *sources.KafkaSource, secret *corev1.Secret) (*contract.Resource, error) {
+	destination, err := r.Resolver.URIFromDestinationV1(ctx, ks.Spec.Sink, ks)
+	if err != nil {
+		return nil, fmt.Errorf("failed to resolve destination: %w", err)
+	}
+
+	egress := &contract.Egress{
+		ConsumerGroup: ks.Spec.ConsumerGroup,
+		Destination:   destination.String(),
+		Uid:           string(ks.GetUID()),
+		EgressConfig: &contract.EgressConfig{
+			// TODO hardcoded retry num
+			Retry:         10,
+			BackoffPolicy: contract.BackoffPolicy_Exponential,
+			// TODO hardcoded backoff delay
+			BackoffDelay: 10000, // 10 seconds
+			// TODO set timeout
+			Timeout: 0,
+		},
+		// TODO delivery ordering
+		DeliveryOrder: contract.DeliveryOrder_ORDERED,
+	}
+	resource := &contract.Resource{
+		Uid:              string(ks.GetUID()),
+		Topics:           ks.Spec.Topics,
+		BootstrapServers: strings.Join(ks.Spec.BootstrapServers, ","),
+		Egresses:         []*contract.Egress{egress},
+		Auth:             &contract.Resource_AbsentAuth{},
+	}
+	if secret != nil {
+		resource.Auth = &contract.Resource_AuthSecret{
+			AuthSecret: &contract.Reference{
+				Uuid:      string(secret.GetUID()),
+				Namespace: secret.GetNamespace(),
+				Name:      secret.GetName(),
+				Version:   secret.GetResourceVersion(),
+			},
+		}
+	}
+	return resource, nil
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index cdb526178..4915a2bfb 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -92,7 +92,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 	}
 
 	// Ignore Triggers that are associated with a Broker we don't own.
-	if isOur, brokerClass := isOurBroker(broker); !isOur {
+	if isKnativeKafkaBroker, brokerClass := isKnativeKafkaBroker(broker); !isKnativeKafkaBroker {
 		logger.Debug("Ignoring Trigger", zap.String(eventing.BrokerClassAnnotationKey, brokerClass))
 		return nil
 	}
@@ -317,7 +317,7 @@ func deleteTrigger(egresses []*contract.Egress, index int) []*contract.Egress {
 	return egresses[:len(egresses)-1]
 }
 
-func isOurBroker(broker *eventing.Broker) (bool, string) {
+func isKnativeKafkaBroker(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
 	return brokerClass == kafka.BrokerClass, brokerClass
 }
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index df8f7275b..d8d6d7204 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 155 third-party dependencies.
+Lists of 151 third-party dependencies.
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.6 - http://logback.qos.ch/logback-classic)
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.6 - http://logback.qos.ch/logback-core)
      (The Apache Software License, Version 2.0) Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.11.4 - http://github.com/FasterXML/jackson)
@@ -96,10 +96,6 @@ Lists of 155 third-party dependencies.
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.6.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.10.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.10.0 - http://github.com/prometheus/client_java/simpleclient_common)
-     (The Apache Software License, Version 2.0) Sundrio :: Annotations :: Builder (io.sundr:builder-annotations:0.13.1 - https://github.com/sundrio/sundrio/annotations/builder-annotations)
-     (The Apache Software License, Version 2.0) Sundrio :: Annotations :: Resourcecify (io.sundr:resourcecify-annotations:0.13.1 - https://github.com/sundrio/sundrio/annotations/resourcecify-annotations)
-     (The Apache Software License, Version 2.0) Sundrio :: Code generation (io.sundr:sundr-codegen:0.13.1 - https://github.com/sundrio/sundrio/sundr-codegen)
-     (The Apache Software License, Version 2.0) Sundrio :: Core (io.sundr:sundr-core:0.13.1 - https://github.com/sundrio/sundrio/sundr-core)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-auth-common (io.vertx:vertx-auth-common:4.1.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-auth/vertx-auth-common)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-circuit-breaker (io.vertx:vertx-circuit-breaker:4.1.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-circuit-breaker)
      (Eclipse Public License - v 2.0) (The Apache Software License, Version 2.0) Vert.x Core (io.vertx:vertx-core:4.1.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-core)
diff --git a/data-plane/core/pom.xml b/data-plane/core/pom.xml
index 030e359f9..55c87a149 100644
--- a/data-plane/core/pom.xml
+++ b/data-plane/core/pom.xml
@@ -92,6 +92,10 @@
           <groupId>com.fasterxml.jackson.core</groupId>
           <artifactId>jackson-core</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>org.osgi</groupId>
+          <artifactId>org.osgi.core</artifactId>
+        </exclusion>
       </exclusions>
     </dependency>
     <dependency>
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 98441ac3c..f39d4cba8 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -36,7 +36,7 @@
     <maven.shade.plugin.version>3.2.4</maven.shade.plugin.version>
     <maven.editorconfig.plugin.version>0.1.1</maven.editorconfig.plugin.version>
     <maven.jacoco.plugin.version>0.8.7</maven.jacoco.plugin.version>
-    <maven.enforcer.plugin.version>3.0.0-M3</maven.enforcer.plugin.version>
+    <maven.enforcer.plugin.version>3.0.0</maven.enforcer.plugin.version>
     <maven.license.plugin.version>2.0.0</maven.license.plugin.version>
     <mycila.license.plugin.version>4.1</mycila.license.plugin.version>
 
@@ -166,6 +166,10 @@
             <artifactId>okhttp</artifactId>
             <groupId>com.squareup.okhttp3</groupId>
           </exclusion>
+          <exclusion>
+            <artifactId>builder-annotations</artifactId>
+            <groupId>io.sundr</groupId>
+          </exclusion>
         </exclusions>
       </dependency>
 
diff --git a/go.mod b/go.mod
index d310470ab..f031d8c19 100644
--- a/go.mod
+++ b/go.mod
@@ -21,8 +21,8 @@ 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.20211006164845-481fd3e3331b
-	knative.dev/eventing-kafka v0.26.1-0.20211005035853-9fe562d91403
+	knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0
+	knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11
 	knative.dev/hack v0.0.0-20210806075220-815cd312d65c
 	knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97
 	knative.dev/reconciler-test v0.0.0-20210930064245-45904ca4383d
diff --git a/go.sum b/go.sum
index 6c85e98b6..49e5f5722 100644
--- a/go.sum
+++ b/go.sum
@@ -642,7 +642,6 @@ github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6So
 github.com/rogpeppe/fastuuid v1.2.0 h1:Ppwyp6VYCF1nvBTXL3trRso7mXMlRrw9ooo375wvi2s=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
-github.com/rs/dnscache v0.0.0-20210201191234-295bba877686/go.mod h1:qe5TWALJ8/a1Lqznoc5BDHpYX/8HU60Hm2AwRmqzxqA=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
@@ -1302,17 +1301,15 @@ k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKU
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210929151023-4b73baddb22a/go.mod h1:RxOuZnxwP51dsuMxFVXFK+7109d3YUqCmAKFH9n+mG4=
 knative.dev/eventing v0.26.1-0.20210930102845-5b9ac948cefc/go.mod h1:HwxfiCneWF5L3AeDFuTWnEs/aFZqr/pdVWV9i5AH5vs=
-knative.dev/eventing v0.26.1-0.20211006164845-481fd3e3331b h1:Mm+wWAHd+m8GG14SxEZvu96Ad9jDjmqE16DvM0u7Ink=
-knative.dev/eventing v0.26.1-0.20211006164845-481fd3e3331b/go.mod h1:Bv9ByJlbDLOlpwWHISGMum9PnZ2yfQjcOTgcfckWF1M=
-knative.dev/eventing-kafka v0.26.1-0.20211005035853-9fe562d91403 h1:iqgzQxskz0a8kqhgjm7gpp8LW2av5yzTmMcs39Y9N+Y=
-knative.dev/eventing-kafka v0.26.1-0.20211005035853-9fe562d91403/go.mod h1:qK71ngnAY7DRJj7wb1pbW0cCKW5sbAn7fV14KaqP+OY=
+knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0 h1:HOct4MnEEnB8luibpEkGowzpywIRZl9R0EKOBXpI4X4=
+knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0/go.mod h1:YwdAAjLecPcaZkpyaNH3Ajn7kCzsfsvMGYS2hyC+R+I=
+knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11 h1:VCpDGwW6ychG+OcwJtZzh9CCZ5sa+T3M/7JjSBOpbAg=
+knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11/go.mod h1:+NeJwjvTos9goI5LshNCNJmSC9dNHCP8rsh2QroeR8Y=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c h1:nOXoDWAAItwr4o0dp3nHr6skgpVD4IvME/UX84YNl5k=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210806075220-815cd312d65c/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/networking v0.0.0-20210929162523-749575ef53f8/go.mod h1:IQMjYdQwtOsqJ+fQDs8MZGB/cPxXdMc4bpP7fdtXp2Q=
 knative.dev/pkg v0.0.0-20210914164111-4857ab6939e3/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
 knative.dev/pkg v0.0.0-20210929111822-2267a4cbebb8/go.mod h1:r27D20afKNeK+9aNOg+0qMv8JgQcyeP+CAYQIR1jEQY=
-knative.dev/pkg v0.0.0-20211005203644-a00ba487121e/go.mod h1:r27D20afKNeK+9aNOg+0qMv8JgQcyeP+CAYQIR1jEQY=
 knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97 h1:art4USpwHkiBKwu1C2X2KVHpDcH9L3mhGlfrHnQowS4=
 knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97/go.mod h1:r27D20afKNeK+9aNOg+0qMv8JgQcyeP+CAYQIR1jEQY=
 knative.dev/reconciler-test v0.0.0-20210915181908-49fac7555086/go.mod h1:6yDmb26SINSmgw6wVy9qQwgRMewiW8ddkkwGLR0ZvOY=
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go
new file mode 100644
index 000000000..d78c7a786
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go
@@ -0,0 +1,103 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/discovery"
+	fakediscovery "k8s.io/client-go/discovery/fake"
+	"k8s.io/client-go/testing"
+	clientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1"
+	fakebindingsv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1"
+	fakekafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake"
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1"
+	fakemessagingv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1"
+	fakesourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake"
+)
+
+// NewSimpleClientset returns a clientset that will respond with the provided objects.
+// It's backed by a very simple object tracker that processes creates, updates and deletions as-is,
+// without applying any validations and/or defaults. It shouldn't be considered a replacement
+// for a real clientset and is mostly useful in simple unit tests.
+func NewSimpleClientset(objects ...runtime.Object) *Clientset {
+	o := testing.NewObjectTracker(scheme, codecs.UniversalDecoder())
+	for _, obj := range objects {
+		if err := o.Add(obj); err != nil {
+			panic(err)
+		}
+	}
+
+	cs := &Clientset{tracker: o}
+	cs.discovery = &fakediscovery.FakeDiscovery{Fake: &cs.Fake}
+	cs.AddReactor("*", "*", testing.ObjectReaction(o))
+	cs.AddWatchReactor("*", func(action testing.Action) (handled bool, ret watch.Interface, err error) {
+		gvr := action.GetResource()
+		ns := action.GetNamespace()
+		watch, err := o.Watch(gvr, ns)
+		if err != nil {
+			return false, nil, err
+		}
+		return true, watch, nil
+	})
+
+	return cs
+}
+
+// Clientset implements clientset.Interface. Meant to be embedded into a
+// struct to get a default implementation. This makes faking out just the method
+// you want to test easier.
+type Clientset struct {
+	testing.Fake
+	discovery *fakediscovery.FakeDiscovery
+	tracker   testing.ObjectTracker
+}
+
+func (c *Clientset) Discovery() discovery.DiscoveryInterface {
+	return c.discovery
+}
+
+func (c *Clientset) Tracker() testing.ObjectTracker {
+	return c.tracker
+}
+
+var _ clientset.Interface = &Clientset{}
+
+// BindingsV1beta1 retrieves the BindingsV1beta1Client
+func (c *Clientset) BindingsV1beta1() bindingsv1beta1.BindingsV1beta1Interface {
+	return &fakebindingsv1beta1.FakeBindingsV1beta1{Fake: &c.Fake}
+}
+
+// KafkaV1alpha1 retrieves the KafkaV1alpha1Client
+func (c *Clientset) KafkaV1alpha1() kafkav1alpha1.KafkaV1alpha1Interface {
+	return &fakekafkav1alpha1.FakeKafkaV1alpha1{Fake: &c.Fake}
+}
+
+// MessagingV1beta1 retrieves the MessagingV1beta1Client
+func (c *Clientset) MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface {
+	return &fakemessagingv1beta1.FakeMessagingV1beta1{Fake: &c.Fake}
+}
+
+// SourcesV1beta1 retrieves the SourcesV1beta1Client
+func (c *Clientset) SourcesV1beta1() sourcesv1beta1.SourcesV1beta1Interface {
+	return &fakesourcesv1beta1.FakeSourcesV1beta1{Fake: &c.Fake}
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/doc.go
new file mode 100644
index 000000000..3b00d536c
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// This package has the automatically generated fake clientset.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/register.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/register.go
new file mode 100644
index 000000000..a396f69d6
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/register.go
@@ -0,0 +1,62 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	serializer "k8s.io/apimachinery/pkg/runtime/serializer"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+)
+
+var scheme = runtime.NewScheme()
+var codecs = serializer.NewCodecFactory(scheme)
+
+var localSchemeBuilder = runtime.SchemeBuilder{
+	bindingsv1beta1.AddToScheme,
+	kafkav1alpha1.AddToScheme,
+	messagingv1beta1.AddToScheme,
+	sourcesv1beta1.AddToScheme,
+}
+
+// AddToScheme adds all types of this clientset into the given scheme. This allows composition
+// of clientsets, like in:
+//
+//   import (
+//     "k8s.io/client-go/kubernetes"
+//     clientsetscheme "k8s.io/client-go/kubernetes/scheme"
+//     aggregatorclientsetscheme "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/scheme"
+//   )
+//
+//   kclientset, _ := kubernetes.NewForConfig(c)
+//   _ = aggregatorclientsetscheme.AddToScheme(clientsetscheme.Scheme)
+//
+// After this, RawExtensions in Kubernetes types will serialize kube-aggregator types
+// correctly.
+var AddToScheme = localSchemeBuilder.AddToScheme
+
+func init() {
+	v1.AddToGroupVersion(scheme, schema.GroupVersion{Version: "v1"})
+	utilruntime.Must(AddToScheme(scheme))
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_bindings_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_bindings_client.go
new file mode 100644
index 000000000..2237e7bfa
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_bindings_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1"
+)
+
+type FakeBindingsV1beta1 struct {
+	*testing.Fake
+}
+
+func (c *FakeBindingsV1beta1) KafkaBindings(namespace string) v1beta1.KafkaBindingInterface {
+	return &FakeKafkaBindings{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeBindingsV1beta1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_kafkabinding.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_kafkabinding.go
new file mode 100644
index 000000000..609547ae2
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_kafkabinding.go
@@ -0,0 +1,142 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+)
+
+// FakeKafkaBindings implements KafkaBindingInterface
+type FakeKafkaBindings struct {
+	Fake *FakeBindingsV1beta1
+	ns   string
+}
+
+var kafkabindingsResource = schema.GroupVersionResource{Group: "bindings.knative.dev", Version: "v1beta1", Resource: "kafkabindings"}
+
+var kafkabindingsKind = schema.GroupVersionKind{Group: "bindings.knative.dev", Version: "v1beta1", Kind: "KafkaBinding"}
+
+// Get takes name of the kafkaBinding, and returns the corresponding kafkaBinding object, and an error if there is any.
+func (c *FakeKafkaBindings) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(kafkabindingsResource, c.ns, name), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// List takes label and field selectors, and returns the list of KafkaBindings that match those selectors.
+func (c *FakeKafkaBindings) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaBindingList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(kafkabindingsResource, kafkabindingsKind, c.ns, opts), &v1beta1.KafkaBindingList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.KafkaBindingList{ListMeta: obj.(*v1beta1.KafkaBindingList).ListMeta}
+	for _, item := range obj.(*v1beta1.KafkaBindingList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaBindings.
+func (c *FakeKafkaBindings) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(kafkabindingsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a kafkaBinding and creates it.  Returns the server's representation of the kafkaBinding, and an error, if there is any.
+func (c *FakeKafkaBindings) Create(ctx context.Context, kafkaBinding *v1beta1.KafkaBinding, opts v1.CreateOptions) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(kafkabindingsResource, c.ns, kafkaBinding), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// Update takes the representation of a kafkaBinding and updates it. Returns the server's representation of the kafkaBinding, and an error, if there is any.
+func (c *FakeKafkaBindings) Update(ctx context.Context, kafkaBinding *v1beta1.KafkaBinding, opts v1.UpdateOptions) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(kafkabindingsResource, c.ns, kafkaBinding), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeKafkaBindings) UpdateStatus(ctx context.Context, kafkaBinding *v1beta1.KafkaBinding, opts v1.UpdateOptions) (*v1beta1.KafkaBinding, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(kafkabindingsResource, "status", c.ns, kafkaBinding), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// Delete takes name of the kafkaBinding and deletes it. Returns an error if one occurs.
+func (c *FakeKafkaBindings) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(kafkabindingsResource, c.ns, name), &v1beta1.KafkaBinding{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeKafkaBindings) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(kafkabindingsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.KafkaBindingList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched kafkaBinding.
+func (c *FakeKafkaBindings) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(kafkabindingsResource, c.ns, name, pt, data, subresources...), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go
new file mode 100644
index 000000000..53377b6b9
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1"
+)
+
+type FakeKafkaV1alpha1 struct {
+	*testing.Fake
+}
+
+func (c *FakeKafkaV1alpha1) ResetOffsets(namespace string) v1alpha1.ResetOffsetInterface {
+	return &FakeResetOffsets{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeKafkaV1alpha1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go
new file mode 100644
index 000000000..af76f8c0b
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go
@@ -0,0 +1,142 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+// FakeResetOffsets implements ResetOffsetInterface
+type FakeResetOffsets struct {
+	Fake *FakeKafkaV1alpha1
+	ns   string
+}
+
+var resetoffsetsResource = schema.GroupVersionResource{Group: "kafka.eventing.knative.dev", Version: "v1alpha1", Resource: "resetoffsets"}
+
+var resetoffsetsKind = schema.GroupVersionKind{Group: "kafka.eventing.knative.dev", Version: "v1alpha1", Kind: "ResetOffset"}
+
+// Get takes name of the resetOffset, and returns the corresponding resetOffset object, and an error if there is any.
+func (c *FakeResetOffsets) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(resetoffsetsResource, c.ns, name), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// List takes label and field selectors, and returns the list of ResetOffsets that match those selectors.
+func (c *FakeResetOffsets) List(ctx context.Context, opts v1.ListOptions) (result *v1alpha1.ResetOffsetList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(resetoffsetsResource, resetoffsetsKind, c.ns, opts), &v1alpha1.ResetOffsetList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1alpha1.ResetOffsetList{ListMeta: obj.(*v1alpha1.ResetOffsetList).ListMeta}
+	for _, item := range obj.(*v1alpha1.ResetOffsetList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested resetOffsets.
+func (c *FakeResetOffsets) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(resetoffsetsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a resetOffset and creates it.  Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *FakeResetOffsets) Create(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.CreateOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(resetoffsetsResource, c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// Update takes the representation of a resetOffset and updates it. Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *FakeResetOffsets) Update(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(resetoffsetsResource, c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeResetOffsets) UpdateStatus(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (*v1alpha1.ResetOffset, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(resetoffsetsResource, "status", c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// Delete takes name of the resetOffset and deletes it. Returns an error if one occurs.
+func (c *FakeResetOffsets) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(resetoffsetsResource, c.ns, name), &v1alpha1.ResetOffset{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeResetOffsets) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(resetoffsetsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1alpha1.ResetOffsetList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched resetOffset.
+func (c *FakeResetOffsets) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(resetoffsetsResource, c.ns, name, pt, data, subresources...), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go
new file mode 100644
index 000000000..5ab4e5a86
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go
@@ -0,0 +1,142 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+)
+
+// FakeKafkaChannels implements KafkaChannelInterface
+type FakeKafkaChannels struct {
+	Fake *FakeMessagingV1beta1
+	ns   string
+}
+
+var kafkachannelsResource = schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1beta1", Resource: "kafkachannels"}
+
+var kafkachannelsKind = schema.GroupVersionKind{Group: "messaging.knative.dev", Version: "v1beta1", Kind: "KafkaChannel"}
+
+// Get takes name of the kafkaChannel, and returns the corresponding kafkaChannel object, and an error if there is any.
+func (c *FakeKafkaChannels) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(kafkachannelsResource, c.ns, name), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// List takes label and field selectors, and returns the list of KafkaChannels that match those selectors.
+func (c *FakeKafkaChannels) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaChannelList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(kafkachannelsResource, kafkachannelsKind, c.ns, opts), &v1beta1.KafkaChannelList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.KafkaChannelList{ListMeta: obj.(*v1beta1.KafkaChannelList).ListMeta}
+	for _, item := range obj.(*v1beta1.KafkaChannelList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaChannels.
+func (c *FakeKafkaChannels) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(kafkachannelsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a kafkaChannel and creates it.  Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *FakeKafkaChannels) Create(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.CreateOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(kafkachannelsResource, c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// Update takes the representation of a kafkaChannel and updates it. Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *FakeKafkaChannels) Update(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(kafkachannelsResource, c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeKafkaChannels) UpdateStatus(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (*v1beta1.KafkaChannel, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(kafkachannelsResource, "status", c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// Delete takes name of the kafkaChannel and deletes it. Returns an error if one occurs.
+func (c *FakeKafkaChannels) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(kafkachannelsResource, c.ns, name), &v1beta1.KafkaChannel{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeKafkaChannels) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(kafkachannelsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.KafkaChannelList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched kafkaChannel.
+func (c *FakeKafkaChannels) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(kafkachannelsResource, c.ns, name, pt, data, subresources...), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go
new file mode 100644
index 000000000..9befbac81
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1"
+)
+
+type FakeMessagingV1beta1 struct {
+	*testing.Fake
+}
+
+func (c *FakeMessagingV1beta1) KafkaChannels(namespace string) v1beta1.KafkaChannelInterface {
+	return &FakeKafkaChannels{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeMessagingV1beta1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go
new file mode 100644
index 000000000..a35a59c17
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go
@@ -0,0 +1,154 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+)
+
+// FakeKafkaSources implements KafkaSourceInterface
+type FakeKafkaSources struct {
+	Fake *FakeSourcesV1beta1
+	ns   string
+}
+
+var kafkasourcesResource = schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1beta1", Resource: "kafkasources"}
+
+var kafkasourcesKind = schema.GroupVersionKind{Group: "sources.knative.dev", Version: "v1beta1", Kind: "KafkaSource"}
+
+// Get takes name of the kafkaSource, and returns the corresponding kafkaSource object, and an error if there is any.
+func (c *FakeKafkaSources) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(kafkasourcesResource, c.ns, name), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// List takes label and field selectors, and returns the list of KafkaSources that match those selectors.
+func (c *FakeKafkaSources) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaSourceList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(kafkasourcesResource, kafkasourcesKind, c.ns, opts), &v1beta1.KafkaSourceList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.KafkaSourceList{ListMeta: obj.(*v1beta1.KafkaSourceList).ListMeta}
+	for _, item := range obj.(*v1beta1.KafkaSourceList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaSources.
+func (c *FakeKafkaSources) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(kafkasourcesResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a kafkaSource and creates it.  Returns the server's representation of the kafkaSource, and an error, if there is any.
+func (c *FakeKafkaSources) Create(ctx context.Context, kafkaSource *v1beta1.KafkaSource, opts v1.CreateOptions) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(kafkasourcesResource, c.ns, kafkaSource), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// Update takes the representation of a kafkaSource and updates it. Returns the server's representation of the kafkaSource, and an error, if there is any.
+func (c *FakeKafkaSources) Update(ctx context.Context, kafkaSource *v1beta1.KafkaSource, opts v1.UpdateOptions) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(kafkasourcesResource, c.ns, kafkaSource), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeKafkaSources) UpdateStatus(ctx context.Context, kafkaSource *v1beta1.KafkaSource, opts v1.UpdateOptions) (*v1beta1.KafkaSource, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(kafkasourcesResource, "status", c.ns, kafkaSource), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// Delete takes name of the kafkaSource and deletes it. Returns an error if one occurs.
+func (c *FakeKafkaSources) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(kafkasourcesResource, c.ns, name), &v1beta1.KafkaSource{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeKafkaSources) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(kafkasourcesResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.KafkaSourceList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched kafkaSource.
+func (c *FakeKafkaSources) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(kafkasourcesResource, c.ns, name, pt, data, subresources...), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// GetScale takes name of the kafkaSource, and returns the corresponding scale object, and an error if there is any.
+func (c *FakeKafkaSources) GetScale(ctx context.Context, kafkaSourceName string, options v1.GetOptions) (result *autoscalingv1.Scale, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetSubresourceAction(kafkasourcesResource, c.ns, "scale", kafkaSourceName), &autoscalingv1.Scale{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*autoscalingv1.Scale), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_sources_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_sources_client.go
new file mode 100644
index 000000000..e3876454b
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_sources_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1"
+)
+
+type FakeSourcesV1beta1 struct {
+	*testing.Fake
+}
+
+func (c *FakeSourcesV1beta1) KafkaSources(namespace string) v1beta1.KafkaSourceInterface {
+	return &FakeKafkaSources{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeSourcesV1beta1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/client/fake/fake.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/client/fake/fake.go
new file mode 100644
index 000000000..96faa7fc2
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/client/fake/fake.go
@@ -0,0 +1,57 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	rest "k8s.io/client-go/rest"
+	fake "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Fake.RegisterClient(withClient)
+	injection.Fake.RegisterClientFetcher(func(ctx context.Context) interface{} {
+		return Get(ctx)
+	})
+}
+
+func withClient(ctx context.Context, cfg *rest.Config) context.Context {
+	ctx, _ = With(ctx)
+	return ctx
+}
+
+func With(ctx context.Context, objects ...runtime.Object) (context.Context, *fake.Clientset) {
+	cs := fake.NewSimpleClientset(objects...)
+	return context.WithValue(ctx, client.Key{}, cs), cs
+}
+
+// Get extracts the Kubernetes client from the context.
+func Get(ctx context.Context) *fake.Clientset {
+	untyped := ctx.Value(client.Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake.Clientset from context.")
+	}
+	return untyped.(*fake.Clientset)
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake/fake.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake/fake.go
new file mode 100644
index 000000000..be34de9ef
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake/fake.go
@@ -0,0 +1,45 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	externalversions "knative.dev/eventing-kafka/pkg/client/informers/externalversions"
+	fake "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	factory "knative.dev/eventing-kafka/pkg/client/injection/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = factory.Get
+
+func init() {
+	injection.Fake.RegisterInformerFactory(withInformerFactory)
+}
+
+func withInformerFactory(ctx context.Context) context.Context {
+	c := fake.Get(ctx)
+	opts := make([]externalversions.SharedInformerOption, 0, 1)
+	if injection.HasNamespaceScope(ctx) {
+		opts = append(opts, externalversions.WithNamespace(injection.GetNamespaceScope(ctx)))
+	}
+	return context.WithValue(ctx, factory.Key{},
+		externalversions.NewSharedInformerFactoryWithOptions(c, controller.GetResyncPeriod(ctx), opts...))
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake/fake.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake/fake.go
new file mode 100644
index 000000000..80d1e5b6d
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake"
+	kafkasource "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = kafkasource.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Sources().V1beta1().KafkaSources()
+	return context.WithValue(ctx, kafkasource.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/kafkasource.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/kafkasource.go
new file mode 100644
index 000000000..5ee3d8632
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/kafkasource.go
@@ -0,0 +1,106 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apissourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	versioned "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	factory "knative.dev/eventing-kafka/pkg/client/injection/informers/factory"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Sources().V1beta1().KafkaSources()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1beta1.KafkaSourceInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1.KafkaSourceInformer from context.")
+	}
+	return untyped.(v1beta1.KafkaSourceInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+}
+
+var _ v1beta1.KafkaSourceInformer = (*wrapper)(nil)
+var _ sourcesv1beta1.KafkaSourceLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apissourcesv1beta1.KafkaSource{}, 0, nil)
+}
+
+func (w *wrapper) Lister() sourcesv1beta1.KafkaSourceLister {
+	return w
+}
+
+func (w *wrapper) KafkaSources(namespace string) sourcesv1beta1.KafkaSourceNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace}
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apissourcesv1beta1.KafkaSource, err error) {
+	lo, err := w.client.SourcesV1beta1().KafkaSources(w.namespace).List(context.TODO(), v1.ListOptions{
+		LabelSelector: selector.String(),
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apissourcesv1beta1.KafkaSource, error) {
+	return w.client.SourcesV1beta1().KafkaSources(w.namespace).Get(context.TODO(), name, v1.GetOptions{
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
new file mode 100644
index 000000000..93bcb825f
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
@@ -0,0 +1,162 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	context "context"
+	fmt "fmt"
+	reflect "reflect"
+	strings "strings"
+
+	zap "go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	scheme "k8s.io/client-go/kubernetes/scheme"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	record "k8s.io/client-go/tools/record"
+	versionedscheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	kafkasource "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	controller "knative.dev/pkg/controller"
+	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+const (
+	defaultControllerAgentName = "kafkasource-controller"
+	defaultFinalizerName       = "kafkasources.sources.knative.dev"
+)
+
+// NewImpl returns a controller.Impl that handles queuing and feeding work from
+// the queue through an implementation of controller.Reconciler, delegating to
+// the provided Interface and optional Finalizer methods. OptionsFn is used to return
+// controller.ControllerOptions to be used by the internal reconciler.
+func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsFn) *controller.Impl {
+	logger := logging.FromContext(ctx)
+
+	// Check the options function input. It should be 0 or 1.
+	if len(optionsFns) > 1 {
+		logger.Fatal("Up to one options function is supported, found: ", len(optionsFns))
+	}
+
+	kafkasourceInformer := kafkasource.Get(ctx)
+
+	lister := kafkasourceInformer.Lister()
+
+	var promoteFilterFunc func(obj interface{}) bool
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					if promoteFilterFunc != nil {
+						if ok := promoteFilterFunc(elt); !ok {
+							continue
+						}
+					}
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client.Get(ctx),
+		Lister:        lister,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
+
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "sources.knative.dev.KafkaSource"),
+	)
+
+	impl := controller.NewContext(ctx, rec, controller.ControllerOptions{WorkQueueName: ctrTypeName, Logger: logger})
+	agentName := defaultControllerAgentName
+
+	// Pass impl to the options. Save any optional results.
+	for _, fn := range optionsFns {
+		opts := fn(impl)
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.AgentName != "" {
+			agentName = opts.AgentName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+		if opts.PromoteFilterFunc != nil {
+			promoteFilterFunc = opts.PromoteFilterFunc
+		}
+	}
+
+	rec.Recorder = createRecorder(ctx, agentName)
+
+	return impl
+}
+
+func createRecorder(ctx context.Context, agentName string) record.EventRecorder {
+	logger := logging.FromContext(ctx)
+
+	recorder := controller.GetEventRecorder(ctx)
+	if recorder == nil {
+		// Create event broadcaster
+		logger.Debug("Creating event broadcaster")
+		eventBroadcaster := record.NewBroadcaster()
+		watches := []watch.Interface{
+			eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Infof),
+			eventBroadcaster.StartRecordingToSink(
+				&v1.EventSinkImpl{Interface: kubeclient.Get(ctx).CoreV1().Events("")}),
+		}
+		recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: agentName})
+		go func() {
+			<-ctx.Done()
+			for _, w := range watches {
+				w.Stop()
+			}
+		}()
+	}
+
+	return recorder
+}
+
+func init() {
+	versionedscheme.AddToScheme(scheme.Scheme)
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go
new file mode 100644
index 000000000..268361e94
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go
@@ -0,0 +1,450 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	context "context"
+	json "encoding/json"
+	fmt "fmt"
+
+	zap "go.uber.org/zap"
+	v1 "k8s.io/api/core/v1"
+	equality "k8s.io/apimachinery/pkg/api/equality"
+	errors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	sets "k8s.io/apimachinery/pkg/util/sets"
+	record "k8s.io/client-go/tools/record"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	versioned "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
+	controller "knative.dev/pkg/controller"
+	kmp "knative.dev/pkg/kmp"
+	logging "knative.dev/pkg/logging"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// Interface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1beta1.KafkaSource.
+type Interface interface {
+	// ReconcileKind implements custom logic to reconcile v1beta1.KafkaSource. Any changes
+	// to the objects .Status or .Finalizers will be propagated to the stored
+	// object. It is recommended that implementors do not call any update calls
+	// for the Kind inside of ReconcileKind, it is the responsibility of the calling
+	// controller to propagate those properties. The resource passed to ReconcileKind
+	// will always have an empty deletion timestamp.
+	ReconcileKind(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+}
+
+// Finalizer defines the strongly typed interfaces to be implemented by a
+// controller finalizing v1beta1.KafkaSource.
+type Finalizer interface {
+	// FinalizeKind implements custom logic to finalize v1beta1.KafkaSource. Any changes
+	// to the objects .Status or .Finalizers will be ignored. Returning a nil or
+	// Normal type reconciler.Event will allow the finalizer to be deleted on
+	// the resource. The resource passed to FinalizeKind will always have a set
+	// deletion timestamp.
+	FinalizeKind(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+}
+
+// ReadOnlyInterface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1beta1.KafkaSource if they want to process resources for which
+// they are not the leader.
+type ReadOnlyInterface interface {
+	// ObserveKind implements logic to observe v1beta1.KafkaSource.
+	// This method should not write to the API.
+	ObserveKind(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+}
+
+type doReconcile func(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+
+// reconcilerImpl implements controller.Reconciler for v1beta1.KafkaSource resources.
+type reconcilerImpl struct {
+	// LeaderAwareFuncs is inlined to help us implement reconciler.LeaderAware.
+	reconciler.LeaderAwareFuncs
+
+	// Client is used to write back status updates.
+	Client versioned.Interface
+
+	// Listers index properties about resources.
+	Lister sourcesv1beta1.KafkaSourceLister
+
+	// Recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	Recorder record.EventRecorder
+
+	// configStore allows for decorating a context with config maps.
+	// +optional
+	configStore reconciler.ConfigStore
+
+	// reconciler is the implementation of the business logic of the resource.
+	reconciler Interface
+
+	// finalizerName is the name of the finalizer to reconcile.
+	finalizerName string
+
+	// skipStatusUpdates configures whether or not this reconciler automatically updates
+	// the status of the reconciled resource.
+	skipStatusUpdates bool
+}
+
+// Check that our Reconciler implements controller.Reconciler.
+var _ controller.Reconciler = (*reconcilerImpl)(nil)
+
+// Check that our generated Reconciler is always LeaderAware.
+var _ reconciler.LeaderAware = (*reconcilerImpl)(nil)
+
+func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versioned.Interface, lister sourcesv1beta1.KafkaSourceLister, recorder record.EventRecorder, r Interface, options ...controller.Options) controller.Reconciler {
+	// Check the options function input. It should be 0 or 1.
+	if len(options) > 1 {
+		logger.Fatal("Up to one options struct is supported, found: ", len(options))
+	}
+
+	// Fail fast when users inadvertently implement the other LeaderAware interface.
+	// For the typed reconcilers, Promote shouldn't take any arguments.
+	if _, ok := r.(reconciler.LeaderAware); ok {
+		logger.Fatalf("%T implements the incorrect LeaderAware interface. Promote() should not take an argument as genreconciler handles the enqueuing automatically.", r)
+	}
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					// TODO: Consider letting users specify a filter in options.
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client,
+		Lister:        lister,
+		Recorder:      recorder,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	for _, opts := range options {
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+	}
+
+	return rec
+}
+
+// Reconcile implements controller.Reconciler
+func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
+	logger := logging.FromContext(ctx)
+
+	// Initialize the reconciler state. This will convert the namespace/name
+	// string into a distinct namespace and name, determine if this instance of
+	// the reconciler is the leader, and any additional interfaces implemented
+	// by the reconciler. Returns an error is the resource key is invalid.
+	s, err := newState(key, r)
+	if err != nil {
+		logger.Error("Invalid resource key: ", key)
+		return nil
+	}
+
+	// If we are not the leader, and we don't implement either ReadOnly
+	// observer interfaces, then take a fast-path out.
+	if s.isNotLeaderNorObserver() {
+		return controller.NewSkipKey(key)
+	}
+
+	// If configStore is set, attach the frozen configuration to the context.
+	if r.configStore != nil {
+		ctx = r.configStore.ToContext(ctx)
+	}
+
+	// Add the recorder to context.
+	ctx = controller.WithEventRecorder(ctx, r.Recorder)
+
+	// Get the resource with this namespace/name.
+
+	getter := r.Lister.KafkaSources(s.namespace)
+
+	original, err := getter.Get(s.name)
+
+	if errors.IsNotFound(err) {
+		// The resource may no longer exist, in which case we stop processing and call
+		// the ObserveDeletion handler if appropriate.
+		logger.Debugf("Resource %q no longer exists", key)
+		if del, ok := r.reconciler.(reconciler.OnDeletionInterface); ok {
+			return del.ObserveDeletion(ctx, types.NamespacedName{
+				Namespace: s.namespace,
+				Name:      s.name,
+			})
+		}
+		return nil
+	} else if err != nil {
+		return err
+	}
+
+	// Don't modify the informers copy.
+	resource := original.DeepCopy()
+
+	var reconcileEvent reconciler.Event
+
+	name, do := s.reconcileMethodFor(resource)
+	// Append the target method to the logger.
+	logger = logger.With(zap.String("targetMethod", name))
+	switch name {
+	case reconciler.DoReconcileKind:
+		// Set and update the finalizer on resource if r.reconciler
+		// implements Finalizer.
+		if resource, err = r.setFinalizerIfFinalizer(ctx, resource); err != nil {
+			return fmt.Errorf("failed to set finalizers: %w", err)
+		}
+
+		if !r.skipStatusUpdates {
+			reconciler.PreProcessReconcile(ctx, resource)
+		}
+
+		// Reconcile this copy of the resource and then write back any status
+		// updates regardless of whether the reconciliation errored out.
+		reconcileEvent = do(ctx, resource)
+
+		if !r.skipStatusUpdates {
+			reconciler.PostProcessReconcile(ctx, resource, original)
+		}
+
+	case reconciler.DoFinalizeKind:
+		// For finalizing reconcilers, if this resource being marked for deletion
+		// and reconciled cleanly (nil or normal event), remove the finalizer.
+		reconcileEvent = do(ctx, resource)
+
+		if resource, err = r.clearFinalizer(ctx, resource, reconcileEvent); err != nil {
+			return fmt.Errorf("failed to clear finalizers: %w", err)
+		}
+
+	case reconciler.DoObserveKind:
+		// Observe any changes to this resource, since we are not the leader.
+		reconcileEvent = do(ctx, resource)
+
+	}
+
+	// Synchronize the status.
+	switch {
+	case r.skipStatusUpdates:
+		// This reconciler implementation is configured to skip resource updates.
+		// This may mean this reconciler does not observe spec, but reconciles external changes.
+	case equality.Semantic.DeepEqual(original.Status, resource.Status):
+		// If we didn't change anything then don't call updateStatus.
+		// This is important because the copy we loaded from the injectionInformer's
+		// cache may be stale and we don't want to overwrite a prior update
+		// to status with this stale state.
+	case !s.isLeader:
+		// High-availability reconcilers may have many replicas watching the resource, but only
+		// the elected leader is expected to write modifications.
+		logger.Warn("Saw status changes when we aren't the leader!")
+	default:
+		if err = r.updateStatus(ctx, original, resource); err != nil {
+			logger.Warnw("Failed to update resource status", zap.Error(err))
+			r.Recorder.Eventf(resource, v1.EventTypeWarning, "UpdateFailed",
+				"Failed to update status for %q: %v", resource.Name, err)
+			return err
+		}
+	}
+
+	// Report the reconciler event, if any.
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			logger.Infow("Returned an event", zap.Any("event", reconcileEvent))
+			r.Recorder.Event(resource, event.EventType, event.Reason, event.Error())
+
+			// the event was wrapped inside an error, consider the reconciliation as failed
+			if _, isEvent := reconcileEvent.(*reconciler.ReconcilerEvent); !isEvent {
+				return reconcileEvent
+			}
+			return nil
+		}
+
+		if controller.IsSkipKey(reconcileEvent) {
+			// This is a wrapped error, don't emit an event.
+		} else if ok, _ := controller.IsRequeueKey(reconcileEvent); ok {
+			// This is a wrapped error, don't emit an event.
+		} else {
+			logger.Errorw("Returned an error", zap.Error(reconcileEvent))
+			r.Recorder.Event(resource, v1.EventTypeWarning, "InternalError", reconcileEvent.Error())
+		}
+		return reconcileEvent
+	}
+
+	return nil
+}
+
+func (r *reconcilerImpl) updateStatus(ctx context.Context, existing *v1beta1.KafkaSource, desired *v1beta1.KafkaSource) error {
+	existing = existing.DeepCopy()
+	return reconciler.RetryUpdateConflicts(func(attempts int) (err error) {
+		// The first iteration tries to use the injectionInformer's state, subsequent attempts fetch the latest state via API.
+		if attempts > 0 {
+
+			getter := r.Client.SourcesV1beta1().KafkaSources(desired.Namespace)
+
+			existing, err = getter.Get(ctx, desired.Name, metav1.GetOptions{})
+			if err != nil {
+				return err
+			}
+		}
+
+		// If there's nothing to update, just return.
+		if equality.Semantic.DeepEqual(existing.Status, desired.Status) {
+			return nil
+		}
+
+		if diff, err := kmp.SafeDiff(existing.Status, desired.Status); err == nil && diff != "" {
+			logging.FromContext(ctx).Debug("Updating status with: ", diff)
+		}
+
+		existing.Status = desired.Status
+
+		updater := r.Client.SourcesV1beta1().KafkaSources(existing.Namespace)
+
+		_, err = updater.UpdateStatus(ctx, existing, metav1.UpdateOptions{})
+		return err
+	})
+}
+
+// updateFinalizersFiltered will update the Finalizers of the resource.
+// TODO: this method could be generic and sync all finalizers. For now it only
+// updates defaultFinalizerName or its override.
+func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource *v1beta1.KafkaSource) (*v1beta1.KafkaSource, error) {
+
+	getter := r.Lister.KafkaSources(resource.Namespace)
+
+	actual, err := getter.Get(resource.Name)
+	if err != nil {
+		return resource, err
+	}
+
+	// Don't modify the informers copy.
+	existing := actual.DeepCopy()
+
+	var finalizers []string
+
+	// If there's nothing to update, just return.
+	existingFinalizers := sets.NewString(existing.Finalizers...)
+	desiredFinalizers := sets.NewString(resource.Finalizers...)
+
+	if desiredFinalizers.Has(r.finalizerName) {
+		if existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Add the finalizer.
+		finalizers = append(existing.Finalizers, r.finalizerName)
+	} else {
+		if !existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Remove the finalizer.
+		existingFinalizers.Delete(r.finalizerName)
+		finalizers = existingFinalizers.List()
+	}
+
+	mergePatch := map[string]interface{}{
+		"metadata": map[string]interface{}{
+			"finalizers":      finalizers,
+			"resourceVersion": existing.ResourceVersion,
+		},
+	}
+
+	patch, err := json.Marshal(mergePatch)
+	if err != nil {
+		return resource, err
+	}
+
+	patcher := r.Client.SourcesV1beta1().KafkaSources(resource.Namespace)
+
+	resourceName := resource.Name
+	updated, err := patcher.Patch(ctx, resourceName, types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		r.Recorder.Eventf(existing, v1.EventTypeWarning, "FinalizerUpdateFailed",
+			"Failed to update finalizers for %q: %v", resourceName, err)
+	} else {
+		r.Recorder.Eventf(updated, v1.EventTypeNormal, "FinalizerUpdate",
+			"Updated %q finalizers", resource.GetName())
+	}
+	return updated, err
+}
+
+func (r *reconcilerImpl) setFinalizerIfFinalizer(ctx context.Context, resource *v1beta1.KafkaSource) (*v1beta1.KafkaSource, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	// If this resource is not being deleted, mark the finalizer.
+	if resource.GetDeletionTimestamp().IsZero() {
+		finalizers.Insert(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
+
+func (r *reconcilerImpl) clearFinalizer(ctx context.Context, resource *v1beta1.KafkaSource, reconcileEvent reconciler.Event) (*v1beta1.KafkaSource, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+	if resource.GetDeletionTimestamp().IsZero() {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			if event.EventType == v1.EventTypeNormal {
+				finalizers.Delete(r.finalizerName)
+			}
+		}
+	} else {
+		finalizers.Delete(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/state.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/state.go
new file mode 100644
index 000000000..3f5c5c64d
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/state.go
@@ -0,0 +1,97 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	fmt "fmt"
+
+	types "k8s.io/apimachinery/pkg/types"
+	cache "k8s.io/client-go/tools/cache"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// state is used to track the state of a reconciler in a single run.
+type state struct {
+	// key is the original reconciliation key from the queue.
+	key string
+	// namespace is the namespace split from the reconciliation key.
+	namespace string
+	// name is the name split from the reconciliation key.
+	name string
+	// reconciler is the reconciler.
+	reconciler Interface
+	// roi is the read only interface cast of the reconciler.
+	roi ReadOnlyInterface
+	// isROI (Read Only Interface) the reconciler only observes reconciliation.
+	isROI bool
+	// isLeader the instance of the reconciler is the elected leader.
+	isLeader bool
+}
+
+func newState(key string, r *reconcilerImpl) (*state, error) {
+	// Convert the namespace/name string into a distinct namespace and name.
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return nil, fmt.Errorf("invalid resource key: %s", key)
+	}
+
+	roi, isROI := r.reconciler.(ReadOnlyInterface)
+
+	isLeader := r.IsLeaderFor(types.NamespacedName{
+		Namespace: namespace,
+		Name:      name,
+	})
+
+	return &state{
+		key:        key,
+		namespace:  namespace,
+		name:       name,
+		reconciler: r.reconciler,
+		roi:        roi,
+		isROI:      isROI,
+		isLeader:   isLeader,
+	}, nil
+}
+
+// isNotLeaderNorObserver checks to see if this reconciler with the current
+// state is enabled to do any work or not.
+// isNotLeaderNorObserver returns true when there is no work possible for the
+// reconciler.
+func (s *state) isNotLeaderNorObserver() bool {
+	if !s.isLeader && !s.isROI {
+		// If we are not the leader, and we don't implement the ReadOnly
+		// interface, then take a fast-path out.
+		return true
+	}
+	return false
+}
+
+func (s *state) reconcileMethodFor(o *v1beta1.KafkaSource) (string, doReconcile) {
+	if o.GetDeletionTimestamp().IsZero() {
+		if s.isLeader {
+			return reconciler.DoReconcileKind, s.reconciler.ReconcileKind
+		} else if s.isROI {
+			return reconciler.DoObserveKind, s.roi.ObserveKind
+		}
+	} else if fin, ok := s.reconciler.(Finalizer); s.isLeader && ok {
+		return reconciler.DoFinalizeKind, fin.FinalizeKind
+	}
+	return "unknown", nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index e483a421b..bec02ec65 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1028,7 +1028,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.26.1-0.20211006164845-481fd3e3331b
+# knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1136,7 +1136,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.20211005035853-9fe562d91403
+# knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1149,11 +1149,16 @@ knative.dev/eventing-kafka/pkg/apis/sources/config
 knative.dev/eventing-kafka/pkg/apis/sources/v1beta1
 knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants
 knative.dev/eventing-kafka/pkg/client/clientset/versioned
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake
 knative.dev/eventing-kafka/pkg/client/informers/externalversions
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings/v1beta1
@@ -1165,9 +1170,14 @@ knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1bet
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1
 knative.dev/eventing-kafka/pkg/client/injection/client
+knative.dev/eventing-kafka/pkg/client/injection/client/fake
 knative.dev/eventing-kafka/pkg/client/injection/informers/factory
+knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake
 knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel
+knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource
+knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake
 knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel
+knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource
 knative.dev/eventing-kafka/pkg/client/listers/bindings/v1beta1
 knative.dev/eventing-kafka/pkg/client/listers/kafka/v1alpha1
 knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 5de321403..55a8f5481 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -85,7 +85,7 @@ var (
 )
 
 func TestBrokerReconciler(t *testing.T) {
-	eventing.RegisterAlternateBrokerConditionSet(base.ConditionSet)
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index a89531352..8e5365dbc 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -36,7 +36,7 @@ import (
 	. "knative.dev/pkg/reconciler/testing"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing"
-	v1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	fakeeventingkafkaclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
 	sinkreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
@@ -78,7 +78,7 @@ var (
 
 func TestSinkReconciler(t *testing.T) {
 
-	v1alpha1.RegisterConditionSet(base.ConditionSet)
+	v1alpha1.RegisterConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
@@ -821,7 +821,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 
 func TestSinkFinalizer(t *testing.T) {
 
-	v1alpha1.RegisterConditionSet(base.ConditionSet)
+	v1alpha1.RegisterConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
diff --git a/control-plane/pkg/reconciler/source/controller_test.go b/control-plane/pkg/reconciler/source/controller_test.go
new file mode 100644
index 000000000..20dfc6c67
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/controller_test.go
@@ -0,0 +1,76 @@
+/*
+ * 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 source
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	"knative.dev/pkg/configmap"
+	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+)
+
+func TestNewController(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	configs := &config.Env{
+		SystemNamespace:      "cm",
+		GeneralConfigMapName: "cm",
+	}
+
+	ctx, _ = fakekubeclient.With(
+		ctx,
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      configs.GeneralConfigMapName,
+				Namespace: configs.SystemNamespace,
+			},
+		},
+	)
+	dynamicScheme := runtime.NewScheme()
+	_ = fakekubeclientset.AddToScheme(dynamicScheme)
+
+	dynamicclient.With(ctx, dynamicScheme)
+
+	controller := NewController(
+		ctx,
+		configmap.NewStaticWatcher(&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "cm",
+			},
+		}),
+		configs,
+	)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 6733433ea..d30fc0d35 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -77,7 +77,7 @@ var (
 )
 
 func TestTriggerReconciler(t *testing.T) {
-	eventing.RegisterAlternateBrokerConditionSet(base.ConditionSet)
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index d21451ae1..cdb86e904 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -26,10 +26,9 @@ fi
 
 if ! ${SKIP_INITIALIZE}; then
   initialize $@ --skip-istio-addon
+  save_release_artifacts || fail_test "Failed to save release artifacts"
 fi
 
-save_release_artifacts || fail_test "Failed to save release artifacts"
-
 if ! ${LOCAL_DEVELOPMENT}; then
   scale_controlplane kafka-controller kafka-webhook-eventing eventing-webhook eventing-controller
   apply_sacura || fail_test "Failed to apply Sacura"
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index f80de6871..9dc621df0 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20211005-9fe562d91"
+    kafka.eventing.knative.dev/release: "v20211012-a8f8dec33"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 956256dbf..8677448f9 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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -469,7 +469,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -517,8 +517,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "96896b00"
@@ -579,10 +579,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -633,10 +633,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -710,10 +710,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "4002b4c2"
@@ -773,10 +773,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/high-availability: "true"
     app.kubernetes.io/name: eventing-controller
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -786,9 +786,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20211005-a6f2a7093"
+        eventing.knative.dev/release: "v20211010-a046d8d24"
         app.kubernetes.io/name: eventing-autoscaler
-        app.kubernetes.io/version: "20211005-a6f2a7093"
+        app.kubernetes.io/version: "20211010-a046d8d24"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -806,7 +806,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:7180aaef0377c07f317570b4a9d09d308138ca9ac31395528eb73bfadaa33bf3
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:d266ab0b340d1f7da0e396e4df110b04c20e15f14eecb18c4996fd2df70e0ebc
           resources:
             requests:
               cpu: 100m
@@ -824,7 +824,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:37609cc4d28d23b43cc617112c7d1066788d8c3cb44060b903cff410a9288244
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:213f9a928827c6bf165d423c819e0f08058bb8175993da4afc69aa3e8796f176
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -868,9 +868,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     app.kubernetes.io/name: pingsource-mt-adapter
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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.
@@ -883,9 +883,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211005-a6f2a7093"
+        eventing.knative.dev/release: "v20211010-a046d8d24"
         app.kubernetes.io/name: pingsource-mt-adapter
-        app.kubernetes.io/version: "20211005-a6f2a7093"
+        app.kubernetes.io/version: "20211010-a046d8d24"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       affinity:
@@ -899,7 +899,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:a6e184013e1651ec5ec693c388487d19f4a772bd116ca3f6efc03d2f4277f574
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:3607079eddad8a294f691790c1acc1e5320ca091ad8dc24455bd4c7ea7f9fb4e
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -957,9 +957,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   scaleTargetRef:
@@ -983,9 +983,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   minAvailable: 80%
@@ -1014,9 +1014,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -1027,9 +1027,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211005-a6f2a7093"
+        eventing.knative.dev/release: "v20211010-a046d8d24"
         app.kubernetes.io/name: eventing-webhook
-        app.kubernetes.io/version: "20211005-a6f2a7093"
+        app.kubernetes.io/version: "20211010-a046d8d24"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1049,7 +1049,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:f96518365b24169a2c6e840fb59067b61c17f4ea011c8e85399a8cf652c720f2
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:d20e0d6d6427c5230410d75842fe5d32f5905e9ae0b06e182276af0db101bb32
           resources:
             requests:
               # taken from serving.
@@ -1113,10 +1113,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     role: eventing-webhook
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1148,11 +1148,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1364,10 +1364,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1528,11 +1528,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1800,11 +1800,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1954,9 +1954,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2083,10 +2083,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2338,11 +2338,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2529,10 +2529,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2850,12 +2850,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3042,9 +3042,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3254,9 +3254,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3439,8 +3439,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3453,9 +3453,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3473,9 +3473,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3496,9 +3496,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3523,9 +3523,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3544,9 +3544,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: messaging-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3567,9 +3567,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3605,8 +3605,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3632,8 +3632,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3650,8 +3650,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3684,8 +3684,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3698,9 +3698,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3738,9 +3738,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3752,9 +3752,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3766,9 +3766,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3780,9 +3780,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3794,9 +3794,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3809,8 +3809,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3823,8 +3823,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3851,8 +3851,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3987,8 +3987,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4055,8 +4055,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4069,9 +4069,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4117,8 +4117,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4131,9 +4131,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4169,8 +4169,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4256,8 +4256,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4356,8 +4356,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4393,8 +4393,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4431,8 +4431,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4465,8 +4465,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4500,8 +4500,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 # The data is populated at install time.
 
@@ -4525,8 +4525,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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 272efe16f..d567fd2c1 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: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     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: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -397,11 +397,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -669,11 +669,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -823,9 +823,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -952,10 +952,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1207,11 +1207,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1398,10 +1398,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1719,12 +1719,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1911,9 +1911,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2123,9 +2123,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211005-a6f2a7093"
+    eventing.knative.dev/release: "v20211010-a046d8d24"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211005-a6f2a7093"
+    app.kubernetes.io/version: "20211010-a046d8d24"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
index 99ad8d18b..057429301 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
@@ -123,8 +123,13 @@ func (o *Receiver) Start(ctx context.Context, handlerFuncs ...func(handler http.
 	for _, dec := range handlerFuncs {
 		handler = dec(handler)
 	}
+	mux := http.NewServeMux()
+	mux.Handle("/", handler)
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
 
-	server := &http.Server{Addr: ":8080", Handler: handler}
+	server := &http.Server{Addr: fmt.Sprintf(":%d", recordevents.EventRecordReceivePort), Handler: mux}
 
 	var err error
 	go func() {
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
index 62e0a4a89..a1bc4b0ee 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
@@ -23,12 +23,15 @@ import (
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	pkgtest "knative.dev/pkg/test"
 
 	testlib "knative.dev/eventing/test/lib"
 )
 
+const EventRecordReceivePort = 8080
+
 func noop(pod *corev1.Pod, client *testlib.Client) error {
 	return nil
 }
@@ -49,7 +52,8 @@ func DeployEventRecordOrFail(ctx context.Context, client *testlib.Client, name s
 		envOption("EVENT_GENERATORS", "receiver"),
 	)
 
-	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace)
+	readinessProbe := true
+	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace, readinessProbe)
 	client.CreatePodOrFail(eventRecordPod, options...)
 	err := pkgtest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
 	if err != nil {
@@ -67,7 +71,8 @@ func DeployEventSenderOrFail(ctx context.Context, client *testlib.Client, name s
 		envOption("SINK", sink),
 	)
 
-	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace)
+	readinessProbe := false // this is not needed for sender pod
+	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace, readinessProbe)
 	client.CreatePodOrFail(eventRecordPod, options...)
 	err := pkgtest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
 	if err != nil {
@@ -76,8 +81,8 @@ func DeployEventSenderOrFail(ctx context.Context, client *testlib.Client, name s
 	return eventRecordPod
 }
 
-func recordEventsPod(imageName string, name string, serviceAccountName string) *corev1.Pod {
-	return &corev1.Pod{
+func recordEventsPod(imageName string, name string, serviceAccountName string, readinessProbe bool) *corev1.Pod {
+	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   name,
 			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
@@ -101,9 +106,26 @@ func recordEventsPod(imageName string, name string, serviceAccountName string) *
 					Name:  "EVENT_LOGS",
 					Value: "recorder,logger",
 				}},
+				Ports: []corev1.ContainerPort{
+					{
+						Name:          "receive",
+						ContainerPort: EventRecordReceivePort,
+					},
+				},
 			}},
 			ServiceAccountName: serviceAccountName,
 			RestartPolicy:      corev1.RestartPolicyAlways,
 		},
 	}
+	if readinessProbe {
+		pod.Spec.Containers[0].ReadinessProbe = &corev1.Probe{
+			Handler: corev1.Handler{
+				HTTPGet: &corev1.HTTPGetAction{
+					Port: intstr.FromString("receive"),
+					Path: "/healthz",
+				},
+			},
+		}
+	}
+	return pod
 }