Repository: knative-sandbox/eventing-kafka-broker
Pull request #1317 (2021-10-14 11:05:30)

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

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

Details:
```

```


Patch:
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index 63f78dc3c..c866d3a39 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -51,7 +51,6 @@ var EgressConditionSet = apis.NewLivingConditionSet(
 	ConditionDataPlaneAvailable,
 	ConditionTopicReady,
 	ConditionConfigMapUpdated,
-	ConditionConfigParsed,
 )
 
 const (
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index f66dba4b6..397c925a2 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -51,6 +51,9 @@ const (
 	// label for selecting sink receiver pods.
 	SinkReceiverLabel = "kafka-sink-receiver"
 
+	// label for selecting source dipsatcher pods.
+	SourceDispatcherLabel = "kafka-source-dispatcher"
+
 	// label for selecting channel dispatcher pods.
 	ChannelDispatcherLabel = "kafka-channel-dispatcher"
 	// label for selecting channel receiver pods.
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
index 8f06ce8b8..da5e89bc8 100644
--- a/control-plane/pkg/reconciler/source/controller.go
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -48,8 +48,7 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
 			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
 			SystemNamespace:             configs.SystemNamespace,
-			DispatcherLabel:             base.BrokerDispatcherLabel,
-			ReceiverLabel:               base.BrokerReceiverLabel,
+			DispatcherLabel:             base.SourceDispatcherLabel,
 		},
 		Env:          configs,
 		ClusterAdmin: sarama.NewClusterAdmin,
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 49d15af90..a150b89c2 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"go.uber.org/zap"
+	"google.golang.org/protobuf/proto"
 	corev1 "k8s.io/api/core/v1"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	"knative.dev/pkg/controller"
@@ -37,6 +38,19 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
+var (
+	DefaultEgressConfig = contract.EgressConfig{
+		Retry:         10,
+		BackoffPolicy: contract.BackoffPolicy_Exponential,
+		BackoffDelay:  10000, // 10 seconds
+		Timeout:       0,
+	}
+)
+
+const (
+	DefaultDeliveryOrder = contract.DeliveryOrder_ORDERED
+)
+
 type Reconciler struct {
 	*base.Reconciler
 	*config.Env
@@ -58,6 +72,11 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		Recorder: controller.GetEventRecorder(ctx),
 	}
 
+	if !r.IsDispatcherRunning() {
+		return statusConditionManager.DataPlaneNotAvailable()
+	}
+	statusConditionManager.DataPlaneAvailable()
+
 	secret, err := security.Secret(ctx, &SecretLocator{KafkaSource: ks}, r.SecretProviderFunc())
 	if err != nil {
 		return fmt.Errorf("failed to get secret: %w", err)
@@ -84,6 +103,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource)
 			return fmt.Errorf("failed to verify topic validity: %w", err)
 		}
 	}
+	statusConditionManager.TopicReady(strings.Join(ks.Spec.Topics, ", "))
 
 	// Get contract config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
@@ -156,21 +176,14 @@ func (r *Reconciler) getResource(ctx context.Context, ks *sources.KafkaSource, s
 		return nil, fmt.Errorf("failed to resolve destination: %w", err)
 	}
 
+	egressConfig := proto.Clone(&DefaultEgressConfig).(*contract.EgressConfig)
+
 	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,
+		EgressConfig:  egressConfig,
+		DeliveryOrder: DefaultDeliveryOrder,
 	}
 	resource := &contract.Resource{
 		Uid:              string(ks.GetUID()),
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index d8d6d7204..203c9bb08 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 151 third-party dependencies.
+Lists of 153 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)
@@ -10,7 +10,7 @@ Lists of 151 third-party dependencies.
      (The Apache Software License, Version 2.0) Jackson datatype: jdk8 (com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.12.5 - https://github.com/FasterXML/jackson-modules-java8/jackson-datatype-jdk8)
      (The Apache Software License, Version 2.0) Jackson datatype: JSR310 (com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.12.5 - https://github.com/FasterXML/jackson-modules-java8/jackson-datatype-jsr310)
      (The Apache Software License, Version 2.0) jackson-module-scala (com.fasterxml.jackson.module:jackson-module-scala_2.12:2.12.5 - http://wiki.fasterxml.com/JacksonModuleScala)
-     (BSD 2-Clause License) zstd-jni (com.github.luben:zstd-jni:1.4.5-6 - https://github.com/luben/zstd-jni)
+     (BSD 2-Clause License) zstd-jni (com.github.luben:zstd-jni:1.4.9-1 - https://github.com/luben/zstd-jni)
      (The Apache Software License, Version 2.0) Generex (com.github.mifmif:generex:1.0.2 - https://github.com/mifmif/Generex/tree/master)
      (The Apache Software License, Version 2.0) FindBugs-jsr305 (com.google.code.findbugs:jsr305:3.0.2 - http://findbugs.sourceforge.net/)
      (Apache 2.0) Gson (com.google.code.gson:gson:2.8.6 - https://github.com/google/gson/gson)
@@ -41,8 +41,8 @@ Lists of 151 third-party dependencies.
      (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
      (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
      (The Apache Software License, Version 2.0) cloudevents-sql (io.cloudevents:cloudevents-sql:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-sql/)
-     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.6.2.Final - https://debezium.io/debezium-parent/debezium-api)
-     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.6.2.Final - https://debezium.io/debezium-parent/debezium-core)
+     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.7.0.Final - https://debezium.io/debezium-parent/debezium-api)
+     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.7.0.Final - https://debezium.io/debezium-parent/debezium-core)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.7.3 - http://fabric8.io/kubernetes-client/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.7.3 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.7.3 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
@@ -114,14 +114,16 @@ Lists of 151 third-party dependencies.
      (Apache License, Version 2.0) (MIT License) Logstash Logback Encoder (net.logstash.logback:logstash-logback-encoder:6.6 - https://github.com/logstash/logstash-logback-encoder)
      (The MIT License) JOpt Simple (net.sf.jopt-simple:jopt-simple:4.6 - http://pholser.github.com/jopt-simple)
      (The MIT License) JOpt Simple (net.sf.jopt-simple:jopt-simple:5.0.4 - http://jopt-simple.github.io/jopt-simple)
+     (MIT) argparse4j (net.sourceforge.argparse4j:argparse4j:0.7.0 - http://argparse4j.github.io)
      (The BSD License) ANTLR 4 Runtime (org.antlr:antlr4-runtime:4.9.2 - http://www.antlr.org/antlr4-runtime)
      (The Apache Software License, Version 2.0) Commons Math (org.apache.commons:commons-math3:3.2 - http://commons.apache.org/proper/commons-math/)
-     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-clients:2.7.0 - https://kafka.apache.org)
-     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-raft:2.7.0 - https://kafka.apache.org)
-     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka_2.12:2.7.0 - https://kafka.apache.org)
+     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-clients:2.8.1 - https://kafka.apache.org)
+     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-metadata:2.8.1 - https://kafka.apache.org)
+     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka-raft:2.8.1 - https://kafka.apache.org)
+     (The Apache Software License, Version 2.0) Apache Kafka (org.apache.kafka:kafka_2.12:2.8.1 - https://kafka.apache.org)
      (Apache License, Version 2.0) Apache Yetus - Audience Annotations (org.apache.yetus:audience-annotations:0.5.0 - https://yetus.apache.org/audience-annotations)
-     (Apache License, Version 2.0) Apache ZooKeeper - Server (org.apache.zookeeper:zookeeper:3.5.8 - http://zookeeper.apache.org/zookeeper)
-     (Apache License, Version 2.0) Apache ZooKeeper - Jute (org.apache.zookeeper:zookeeper-jute:3.5.8 - http://zookeeper.apache.org/zookeeper-jute)
+     (Apache License, Version 2.0) Apache ZooKeeper - Server (org.apache.zookeeper:zookeeper:3.5.9 - http://zookeeper.apache.org/zookeeper)
+     (Apache License, Version 2.0) Apache ZooKeeper - Jute (org.apache.zookeeper:zookeeper-jute:3.5.9 - http://zookeeper.apache.org/zookeeper-jute)
      (The Apache License, Version 2.0) org.apiguardian:apiguardian-api (org.apiguardian:apiguardian-api:1.1.0 - https://github.com/apiguardian-team/apiguardian)
      (Apache License, Version 2.0) AssertJ fluent assertions (org.assertj:assertj-core:3.20.1 - https://assertj.github.io/doc/assertj-core/)
      (Apache 2.0) Awaitility (org.awaitility:awaitility:4.1.0 - http://awaitility.org)
@@ -143,11 +145,11 @@ Lists of 151 third-party dependencies.
      (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Core (org.openjdk.jmh:jmh-core:1.33 - http://openjdk.java.net/projects/code-tools/jmh/jmh-core/)
      (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Generators: Annotation Processors (org.openjdk.jmh:jmh-generator-annprocess:1.33 - http://openjdk.java.net/projects/code-tools/jmh/jmh-generator-annprocess/)
      (The Apache License, Version 2.0) org.opentest4j:opentest4j (org.opentest4j:opentest4j:1.2.0 - https://github.com/ota4j-team/opentest4j)
-     (Apache-2.0) Scala Library (org.scala-lang:scala-library:2.12.12 - https://www.scala-lang.org/)
-     (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.12 - https://www.scala-lang.org/)
-     (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.2.0 - http://www.scala-lang.org/)
+     (Apache-2.0) Scala Library (org.scala-lang:scala-library:2.12.13 - https://www.scala-lang.org/)
+     (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.13 - https://www.scala-lang.org/)
+     (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.3.0 - http://www.scala-lang.org/)
      (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 - http://www.scala-lang.org/)
      (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.32 - http://www.slf4j.org)
      (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.32 - http://www.slf4j.org)
-     (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.7.7 - https://github.com/xerial/snappy-java)
+     (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.8.1 - https://github.com/xerial/snappy-java)
      (Apache License, Version 2.0) SnakeYAML (org.yaml:snakeyaml:1.26 - http://www.snakeyaml.org)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index f39d4cba8..43e9a0b70 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -55,8 +55,8 @@
     <junit.jupiter.version>5.7.2</junit.jupiter.version>
     <mokito.junit.jupiter.version>3.12.4</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.7.3</fabric8.kubernetes.version>
-    <kafka.version>2.7.0</kafka.version>
-    <debezium.version>1.6.2.Final</debezium.version>
+    <kafka.version>2.8.1</kafka.version>
+    <debezium.version>1.7.0.Final</debezium.version>
   </properties>
 
   <modules>
diff --git a/go.mod b/go.mod
index f031d8c19..90830743c 100644
--- a/go.mod
+++ b/go.mod
@@ -21,9 +21,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0
-	knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11
+	knative.dev/eventing v0.26.1-0.20211014072442-a6a819dc71cf
+	knative.dev/eventing-kafka v0.26.1-0.20211014113242-ebe7bfbe9c0d
 	knative.dev/hack v0.0.0-20210806075220-815cd312d65c
-	knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97
+	knative.dev/pkg v0.0.0-20211013152848-fa2f8f19557b
 	knative.dev/reconciler-test v0.0.0-20210930064245-45904ca4383d
 )
diff --git a/go.sum b/go.sum
index 49e5f5722..9e23eadfd 100644
--- a/go.sum
+++ b/go.sum
@@ -1299,20 +1299,19 @@ k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iL
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-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.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/control-protocol v0.0.0-20211013013746-536474c49fa8/go.mod h1:u5+MlOX2uTstj9o/UjIT3Jei/zXgnaAiIUtMBz1uvFM=
+knative.dev/eventing v0.26.1-0.20211013181948-ff4bc408741a/go.mod h1:YwdAAjLecPcaZkpyaNH3Ajn7kCzsfsvMGYS2hyC+R+I=
+knative.dev/eventing v0.26.1-0.20211014072442-a6a819dc71cf h1:p98Lp6JG++VkS5j/N4Z0dXikE0bWKKFVHd/CE8Coe/M=
+knative.dev/eventing v0.26.1-0.20211014072442-a6a819dc71cf/go.mod h1:UnVWSe2+tDUOaGvAtPsszRp23p0zPw5vO/dpranR2mo=
+knative.dev/eventing-kafka v0.26.1-0.20211014113242-ebe7bfbe9c0d h1:DP3jGSdwkC3IAYd9xljuN69L2PkTq1CF8s63PYiSD8Q=
+knative.dev/eventing-kafka v0.26.1-0.20211014113242-ebe7bfbe9c0d/go.mod h1:RHBAhuhX0fVGc1gD6FjLiNq4fRIui36571lZAJP1fhQ=
 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/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-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=
+knative.dev/pkg v0.0.0-20211013152848-fa2f8f19557b h1:k7qT8TtdbyU4koLF745VoBaM5EpUQlV2f13syN8f5L0=
+knative.dev/pkg v0.0.0-20211013152848-fa2f8f19557b/go.mod h1:r27D20afKNeK+9aNOg+0qMv8JgQcyeP+CAYQIR1jEQY=
 knative.dev/reconciler-test v0.0.0-20210930064245-45904ca4383d h1:z0cAefVa6gSOK1wkRKii7YGlrHCuro3/XHbVm00fU3I=
 knative.dev/reconciler-test v0.0.0-20210930064245-45904ca4383d/go.mod h1:6yDmb26SINSmgw6wVy9qQwgRMewiW8ddkkwGLR0ZvOY=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/placement_types.go b/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/placement_types.go
index 34b0e2204..005927291 100644
--- a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/placement_types.go
+++ b/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/placement_types.go
@@ -28,7 +28,7 @@ import (
 // Each pair represents the assignment of virtual replicas to a pod
 type Placeable struct {
 	MaxAllowedVReplicas *int32      `json:"maxAllowedVReplicas,omitempty"`
-	Placement           []Placement `json:"placements,omitempty"`
+	Placements          []Placement `json:"placements,omitempty"`
 }
 
 // PlaceableType is a skeleton type wrapping Placeable in the manner we expect
@@ -67,7 +67,7 @@ func (*Placeable) GetFullType() duck.Populatable {
 
 // Populate implements duck.Populatable
 func (t *PlaceableType) Populate() {
-	t.Status.Placement = []Placement{{PodName: "pod-0", VReplicas: int32(1)}}
+	t.Status.Placements = []Placement{{PodName: "pod-0", VReplicas: int32(1)}}
 }
 
 // GetListType implements apis.Listable
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
index 4a972bfc0..0e0589e5e 100644
--- a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
@@ -32,8 +32,8 @@ func (in *Placeable) DeepCopyInto(out *Placeable) {
 		*out = new(int32)
 		**out = **in
 	}
-	if in.Placement != nil {
-		in, out := &in.Placement, &out.Placement
+	if in.Placements != nil {
+		in, out := &in.Placements, &out.Placements
 		*out = make([]Placement, len(*in))
 		copy(*out, *in)
 	}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_scheduling.go b/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_scheduling.go
index cf0a070db..404b1a90b 100644
--- a/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_scheduling.go
+++ b/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_scheduling.go
@@ -41,10 +41,10 @@ func (k *KafkaSource) GetVReplicas() int32 {
 }
 
 func (k *KafkaSource) GetPlacements() []v1alpha1.Placement {
-	if k.Status.Placeable.Placement == nil {
+	if k.Status.Placeable.Placements == nil {
 		return nil
 	}
-	return k.Status.Placeable.Placement
+	return k.Status.Placeable.Placements
 }
 
 func (k *KafkaSource) GetResourceVersion() string {
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go b/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go
index 1595c8c04..1559d64f9 100644
--- a/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go
@@ -20,4 +20,5 @@ const (
 	KReferenceGroup   = "kreference-group"
 	DeliveryTimeout   = "delivery-timeout"
 	KReferenceMapping = "kreference-mapping"
+	StrictSubscriber  = "strict-subscriber"
 )
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_validation.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_validation.go
index 033654529..7b1b82a06 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_validation.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_validation.go
@@ -19,6 +19,8 @@ package v1
 import (
 	"context"
 
+	"knative.dev/eventing/pkg/apis/feature"
+
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"k8s.io/apimachinery/pkg/api/equality"
 	"knative.dev/pkg/apis"
@@ -50,10 +52,20 @@ func (ss *SubscriptionSpec) Validate(ctx context.Context) *apis.FieldError {
 
 	missingSubscriber := isDestinationNilOrEmpty(ss.Subscriber)
 	missingReply := isDestinationNilOrEmpty(ss.Reply)
-	if missingSubscriber && missingReply {
-		fe := apis.ErrMissingField("reply", "subscriber")
-		fe.Details = "the Subscription must reference at least one of (reply or a subscriber)"
+
+	// Check if StrictSubscriber flag is enabled, if so we follow the spec and check for a valid reference to a subscriber
+	if missingSubscriber && feature.FromContext(ctx).IsEnabled(feature.StrictSubscriber) {
+		fe := apis.ErrMissingField("subscriber")
+		fe.Details = "the Subscription must reference a subscriber"
 		errs = errs.Also(fe)
+
+	} else { // if the flag is not set, we use pre 0.26 behavior
+		if missingSubscriber && missingReply {
+			fe := apis.ErrMissingField("reply", "subscriber")
+			fe.Details = "the Subscription must reference at least one of (reply or a subscriber)"
+			errs = errs.Also(fe)
+		}
+
 	}
 
 	if !missingSubscriber {
diff --git a/vendor/knative.dev/pkg/injection/clients/dynamicclient/fake/fake.go b/vendor/knative.dev/pkg/injection/clients/dynamicclient/fake/fake.go
index aeeb7890a..80aaadd40 100644
--- a/vendor/knative.dev/pkg/injection/clients/dynamicclient/fake/fake.go
+++ b/vendor/knative.dev/pkg/injection/clients/dynamicclient/fake/fake.go
@@ -18,9 +18,7 @@ package fake
 
 import (
 	"context"
-	"strings"
 
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic/fake"
 	k8sscheme "k8s.io/client-go/kubernetes/scheme"
@@ -29,7 +27,6 @@ import (
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/injection/clients/dynamicclient"
 	"knative.dev/pkg/logging"
-	pkgunstructured "knative.dev/pkg/unstructured"
 )
 
 func init() {
@@ -44,39 +41,7 @@ func withClient(ctx context.Context, cfg *rest.Config) context.Context {
 }
 
 func With(ctx context.Context, scheme *runtime.Scheme, objects ...runtime.Object) (context.Context, *fake.FakeDynamicClient) {
-	// We create a scheme were we define all our types and lists
-	// and have them map to unstructured types
-	//
-	// This was a K8s 1.20 breaking change
-	unstructuredScheme := runtime.NewScheme()
-	for gvk := range scheme.AllKnownTypes() {
-		if unstructuredScheme.Recognizes(gvk) {
-			continue
-		}
-		if strings.HasSuffix(gvk.Kind, "List") {
-			unstructuredScheme.AddKnownTypeWithName(gvk, &unstructured.UnstructuredList{})
-			continue
-		}
-		unstructuredScheme.AddKnownTypeWithName(gvk, &unstructured.Unstructured{})
-	}
-
-	objects, err := pkgunstructured.ConvertManyToObjects(scheme, objects)
-	if err != nil {
-		panic(err)
-	}
-
-	for _, obj := range objects {
-		gvk := obj.GetObjectKind().GroupVersionKind()
-		if !unstructuredScheme.Recognizes(gvk) {
-			unstructuredScheme.AddKnownTypeWithName(gvk, &unstructured.Unstructured{})
-		}
-		gvk.Kind += "List"
-		if !unstructuredScheme.Recognizes(gvk) {
-			unstructuredScheme.AddKnownTypeWithName(gvk, &unstructured.UnstructuredList{})
-		}
-	}
-
-	cs := fake.NewSimpleDynamicClient(unstructuredScheme, objects...)
+	cs := fake.NewSimpleDynamicClient(scheme, objects...)
 	return context.WithValue(ctx, dynamicclient.Key{}, cs), cs
 }
 
diff --git a/vendor/knative.dev/pkg/unstructured/unstructured.go b/vendor/knative.dev/pkg/unstructured/unstructured.go
deleted file mode 100644
index cb126c45a..000000000
--- a/vendor/knative.dev/pkg/unstructured/unstructured.go
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package unstructured
-
-import (
-	"fmt"
-
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/runtime"
-)
-
-// ConvertTo converts a runtime.Object to an unstructured.Unstructured type
-func ConvertTo(s *runtime.Scheme, obj runtime.Object) (*unstructured.Unstructured, error) {
-	var (
-		err error
-		u   unstructured.Unstructured
-	)
-
-	u.Object, err = runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
-	if err != nil {
-		return nil, fmt.Errorf("failed to convert to unstructured: %w", err)
-	}
-
-	gvk := u.GroupVersionKind()
-	if gvk.Group == "" || gvk.Kind == "" {
-		gvks, _, err := s.ObjectKinds(obj)
-		if err != nil {
-			return nil, fmt.Errorf("failed to convert to unstructured: %w", err)
-		}
-		apiv, k := gvks[0].ToAPIVersionAndKind()
-		u.SetAPIVersion(apiv)
-		u.SetKind(k)
-	}
-	return &u, nil
-}
-
-// ConvertManyTo converts a slice of runtime.Object to a slice of *unstructured.Unstructured
-func ConvertManyTo(s *runtime.Scheme, objs []runtime.Object) ([]*unstructured.Unstructured, error) {
-	ul := make([]*unstructured.Unstructured, 0, len(objs))
-
-	for _, obj := range objs {
-		u, err := ConvertTo(s, obj)
-		if err != nil {
-			return nil, err
-		}
-
-		ul = append(ul, u)
-	}
-	return ul, nil
-}
-
-// ConvertManyToObjects converts a slice of runtime.Object to a slice of runtime.Objects
-// where each element is of the type *unstructured.Unstructured
-func ConvertManyToObjects(s *runtime.Scheme, objs []runtime.Object) ([]runtime.Object, error) {
-	ul := make([]runtime.Object, 0, len(objs))
-
-	for _, obj := range objs {
-		u, err := ConvertTo(s, obj)
-		if err != nil {
-			return nil, err
-		}
-
-		ul = append(ul, u)
-	}
-	return ul, nil
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index bec02ec65..e2f28965e 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.20211011151816-4bf0a232f3e0
+# knative.dev/eventing v0.26.1-0.20211014072442-a6a819dc71cf
 ## 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.20211011084516-a8f8dec33c11
+# knative.dev/eventing-kafka v0.26.1-0.20211014113242-ebe7bfbe9c0d
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1189,7 +1189,7 @@ knative.dev/eventing-kafka/pkg/common/kafka/sarama
 # knative.dev/hack v0.0.0-20210806075220-815cd312d65c
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97
+# knative.dev/pkg v0.0.0-20211013152848-fa2f8f19557b
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1274,7 +1274,6 @@ knative.dev/pkg/tracing/config
 knative.dev/pkg/tracing/propagation
 knative.dev/pkg/tracing/propagation/tracecontextb3
 knative.dev/pkg/tracker
-knative.dev/pkg/unstructured
 knative.dev/pkg/version
 knative.dev/pkg/webhook
 knative.dev/pkg/webhook/certificates


Test patch:
diff --git a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
index 33633095e..dbc0d04da 100644
--- a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
+++ b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/google/go-cmp/cmp"
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 var _ sarama.ClusterAdmin = &MockKafkaClusterAdmin{}
@@ -77,8 +78,8 @@ func (m *MockKafkaClusterAdmin) ListTopics() (map[string]sarama.TopicDetail, err
 
 func (m *MockKafkaClusterAdmin) DescribeTopics(topics []string) (metadata []*sarama.TopicMetadata, err error) {
 
-	if diff := cmp.Diff(topics, m.ExpectedTopics); diff != "" {
-		m.T.Errorf("unexpected topics (-want, +got) %s", diff)
+	if !sets.NewString(m.ExpectedTopics...).HasAll(topics...) {
+		m.T.Errorf("unexpected topics %v", topics)
 	}
 
 	return m.ExpectedTopicsMetadataOnDescribeTopics, m.ExpectedErrorOnDescribeTopics
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
new file mode 100644
index 000000000..9dab6673c
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -0,0 +1,166 @@
+/*
+ * 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_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+
+	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	eventingkafkasourcereconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/source"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+)
+
+func TestReconcileKind(t *testing.T) {
+
+	sources.RegisterAlternateKafkaConditionSet(base.EgressConditionSet)
+
+	configs := *DefaultConfigs
+	testKey := fmt.Sprintf("%s/%s", SourceNamespace, SourceName)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal - no auth",
+			Objects: []runtime.Object{
+				NewSourceSinkObject(),
+				NewSource(),
+				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              SourceUUID,
+							Topics:           SourceTopics,
+							BootstrapServers: SourceBootstrapServers,
+							Egresses: []*contract.Egress{
+								{
+									ConsumerGroup: SourceConsumerGroup,
+									Destination:   ServiceURL,
+									Uid:           SourceUUID,
+									EgressConfig:  &DefaultEgressConfig,
+									DeliveryOrder: DefaultDeliveryOrder,
+								},
+							},
+							Auth: &contract.Resource_AbsentAuth{},
+						},
+					},
+				}),
+				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMap(&configs, nil),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						SourceConfigMapUpdatedReady(&configs),
+						SourceTopicsReady,
+						SourceDataPlaneAvailable,
+					),
+				},
+			},
+		},
+	}
+
+	table.Test(t, NewFactory(&configs, func(ctx context.Context, listers *Listers, configs *broker.Configs, row *TableRow) controller.Reconciler {
+
+		reconciler := &Reconciler{
+			Reconciler: &base.Reconciler{
+				KubeClient:                  kubeclient.Get(ctx),
+				PodLister:                   listers.GetPodLister(),
+				SecretLister:                listers.GetSecretLister(),
+				DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+				SystemNamespace:             configs.SystemNamespace,
+				DispatcherLabel:             base.SourceDispatcherLabel,
+			},
+			Env: &configs.Env,
+			ClusterAdmin: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopicName:                      "",
+					ExpectedTopicDetail:                    sarama.TopicDetail{},
+					ErrorOnCreateTopic:                     nil,
+					ErrorOnDeleteTopic:                     nil,
+					ExpectedClose:                          false,
+					ExpectedCloseError:                     nil,
+					ExpectedTopics:                         SourceTopics,
+					ExpectedErrorOnDescribeTopics:          nil,
+					ExpectedTopicsMetadataOnDescribeTopics: nil,
+					T:                                      t,
+				}, nil
+			},
+		}
+
+		reconciler.ConfigMapTracker = &FakeTracker{}
+		reconciler.SecretTracker = &FakeTracker{}
+
+		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))
+
+		r := eventingkafkasourcereconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingkafkaclient.Get(ctx),
+			listers.GetKafkaSourceLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+		)
+		return r
+	}))
+}
+
+func SourceDispatcherPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Pod",
+		},
+		namespace,
+		SourceDispatcherPod(namespace, annotations),
+	)
+}
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index e8602ffc5..4339757b6 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -35,7 +35,9 @@ import (
 	"knative.dev/pkg/reconciler"
 	. "knative.dev/pkg/reconciler/testing"
 
-	fakeeventingkafkaclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
+	fakeeventingkafkakafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+
+	fakeeventingkafkabrokerclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
@@ -71,7 +73,8 @@ func NewFactory(configs *broker.Configs, ctor Ctor) Factory {
 		ctx := context.Background()
 
 		ctx, eventingClient := fakeeventingclient.With(ctx, listers.GetEventingObjects()...)
-		ctx, eventingKafkaClient := fakeeventingkafkaclient.With(ctx, listers.GetEventingKafkaObjects()...)
+		ctx, sourcesKafkaClient := fakeeventingkafkakafkaclient.With(ctx, listers.GetEventingKafkaObjects()...)
+		ctx, eventingKafkaClient := fakeeventingkafkabrokerclient.With(ctx, listers.GetEventingKafkaBrokerObjects()...)
 		ctx, kubeClient := fakekubeclient.With(ctx, listers.GetKubeObjects()...)
 
 		ctx, dynamicClient := fakedynamicclient.With(ctx,
@@ -107,6 +110,7 @@ func NewFactory(configs *broker.Configs, ctor Ctor) Factory {
 			dynamicClient.PrependReactor("*", "*", reactor)
 			eventingClient.PrependReactor("*", "*", reactor)
 			eventingKafkaClient.PrependReactor("*", "*", reactor)
+			sourcesKafkaClient.PrependReactor("*", "*", reactor)
 		}
 
 		actionRecorderList := ActionRecorderList{
@@ -114,6 +118,7 @@ func NewFactory(configs *broker.Configs, ctor Ctor) Factory {
 			kubeClient,
 			eventingClient,
 			eventingKafkaClient,
+			sourcesKafkaClient,
 		}
 
 		eventList := EventList{
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 08c14a5df..3c1ff5d4b 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -28,15 +28,20 @@ import (
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	"knative.dev/pkg/reconciler/testing"
 
-	eventingkafka "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-	fakeeventingkafkaclientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/fake"
-	eventingkafkalisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/listers/eventing/v1alpha1"
+	eventingkafkasources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	fakeeventingkafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	eventingkafkasourceslisters "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
+
+	eventingkafkabroker "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	fakeeventingkafkabrokerclientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/fake"
+	eventingkafkabrokerlisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/listers/eventing/v1alpha1"
 )
 
 var clientSetSchemes = []func(*runtime.Scheme) error{
 	fakeeventingclientset.AddToScheme,
 	fakekubeclientset.AddToScheme,
 	fakeapiextensionsclientset.AddToScheme,
+	fakeeventingkafkabrokerclientset.AddToScheme,
 	fakeeventingkafkaclientset.AddToScheme,
 }
 
@@ -80,6 +85,10 @@ func (l *Listers) GetEventingObjects() []runtime.Object {
 	return l.sorter.ObjectsForSchemeFunc(fakeeventingclientset.AddToScheme)
 }
 
+func (l *Listers) GetEventingKafkaBrokerObjects() []runtime.Object {
+	return l.sorter.ObjectsForSchemeFunc(fakeeventingkafkabrokerclientset.AddToScheme)
+}
+
 func (l *Listers) GetEventingKafkaObjects() []runtime.Object {
 	return l.sorter.ObjectsForSchemeFunc(fakeeventingkafkaclientset.AddToScheme)
 }
@@ -104,8 +113,12 @@ func (l *Listers) GetConfigMapLister() corelisters.ConfigMapLister {
 	return corelisters.NewConfigMapLister(l.indexerFor(&corev1.ConfigMap{}))
 }
 
-func (l *Listers) GetKafkaSinkLister() eventingkafkalisters.KafkaSinkLister {
-	return eventingkafkalisters.NewKafkaSinkLister(l.indexerFor(&eventingkafka.KafkaSink{}))
+func (l *Listers) GetKafkaSinkLister() eventingkafkabrokerlisters.KafkaSinkLister {
+	return eventingkafkabrokerlisters.NewKafkaSinkLister(l.indexerFor(&eventingkafkabroker.KafkaSink{}))
+}
+
+func (l *Listers) GetKafkaSourceLister() eventingkafkasourceslisters.KafkaSourceLister {
+	return eventingkafkasourceslisters.NewKafkaSourceLister(l.indexerFor(&eventingkafkasources.KafkaSource{}))
 }
 
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
new file mode 100644
index 000000000..f882a2451
--- /dev/null
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -0,0 +1,129 @@
+/*
+ * 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 testing
+
+import (
+	"fmt"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+)
+
+const (
+	SourceName             = "ks"
+	SourceNamespace        = "test-ns"
+	SourceConsumerGroup    = "ks-group"
+	SourceUUID             = "uuid"
+	SourceBootstrapServers = "kafka:9092"
+)
+
+var (
+	SourceTopics = []string{"t1", "t2"}
+)
+
+type SourceOption func(ks *sources.KafkaSource)
+
+func NewSource(options ...SourceOption) *sources.KafkaSource {
+	s := &sources.KafkaSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: SourceNamespace,
+			Name:      SourceName,
+			UID:       SourceUUID,
+		},
+		Spec: sources.KafkaSourceSpec{
+			KafkaAuthSpec: v1beta1.KafkaAuthSpec{
+				BootstrapServers: []string{SourceBootstrapServers},
+				Net:              v1beta1.KafkaNetSpec{},
+			},
+			Topics:        SourceTopics,
+			ConsumerGroup: SourceConsumerGroup,
+			SourceSpec: duckv1.SourceSpec{
+				Sink: NewSourceSinkReference(),
+			},
+		},
+	}
+	for _, opt := range options {
+		opt(s)
+	}
+	return s
+}
+
+func NewSourceSinkObject() *corev1.Service {
+	return NewService()
+}
+
+func NewSourceSinkReference() duckv1.Destination {
+	s := NewService()
+	return duckv1.Destination{
+		Ref: &duckv1.KReference{
+			Kind:       s.Kind,
+			Namespace:  s.Namespace,
+			Name:       s.Name,
+			APIVersion: s.APIVersion,
+		},
+	}
+}
+
+func SourceConfigMapUpdatedReady(configs *Configs) func(source *sources.KafkaSource) {
+	return func(source *sources.KafkaSource) {
+		source.GetConditionSet().Manage(source.GetStatus()).MarkTrueWithReason(
+			base.ConditionConfigMapUpdated,
+			fmt.Sprintf("Config map %s updated", configs.DataPlaneConfigMapAsString()),
+			"",
+		)
+	}
+}
+
+func SourceTopicsReady(source *sources.KafkaSource) {
+	source.GetConditionSet().Manage(source.GetStatus()).MarkTrueWithReason(
+		base.ConditionTopicReady,
+		fmt.Sprintf("Topic %s created", strings.Join(SourceTopics, ", ")),
+		"",
+	)
+}
+
+func SourceDataPlaneAvailable(source *sources.KafkaSource) {
+	source.GetConditionSet().Manage(source.GetStatus()).MarkTrue(base.ConditionDataPlaneAvailable)
+}
+
+func SourceDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
+	return &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "kafka-source-dispatcher",
+			Namespace:   namespace,
+			Annotations: annotations,
+			Labels: map[string]string{
+				"app": base.SourceDispatcherLabel,
+			},
+		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
+	}
+}
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index 9dc621df0..d171ff2da 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: "v20211012-a8f8dec33"
+    kafka.eventing.knative.dev/release: "v20211013-3a73c372a"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crds.yaml b/third_party/eventing-kafka-latest/source-crds.yaml
new file mode 100644
index 000000000..5f3ba99f1
--- /dev/null
+++ b/third_party/eventing-kafka-latest/source-crds.yaml
@@ -0,0 +1,156 @@
+# Copyright 2019 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  labels:
+    kafka.eventing.knative.dev/release: "v20211013-3a73c372a"
+    duck.knative.dev/binding: "true"
+    knative.dev/crd-install: "true"
+  name: kafkabindings.bindings.knative.dev
+spec:
+  group: bindings.knative.dev
+  versions:
+    - name: v1beta1
+      served: true
+      storage: true
+      subresources:
+        status: {}
+      schema:
+        openAPIV3Schema:
+          type: object
+          # this is a work around so we don't need to flush out the
+          # schema for each version at this time
+          #
+          # see issue: https://github.com/knative/serving/issues/912
+          x-kubernetes-preserve-unknown-fields: true
+      additionalPrinterColumns:
+        - name: BootstrapServers
+          type: string
+          jsonPath: ".spec.bootstrapServers"
+        - name: Ready
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].status"
+        - name: Reason
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].reason"
+        - name: Age
+          type: date
+          jsonPath: .metadata.creationTimestamp
+  names:
+    categories:
+      - all
+      - knative
+      - eventing
+      - bindings
+    kind: KafkaBinding
+    plural: kafkabindings
+  scope: Namespaced
+  conversion:
+    strategy: Webhook
+    webhook:
+      conversionReviewVersions: ["v1", "v1beta1"]
+      clientConfig:
+        service:
+          name: kafka-source-webhook
+          namespace: knative-eventing
+
+---
+# Copyright 2019 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  labels:
+    kafka.eventing.knative.dev/release: "v20211013-3a73c372a"
+    eventing.knative.dev/source: "true"
+    duck.knative.dev/source: "true"
+    knative.dev/crd-install: "true"
+  annotations:
+    registry.knative.dev/eventTypes: |
+      [
+        { "type": "dev.knative.kafka.event" }
+      ]
+  name: kafkasources.sources.knative.dev
+spec:
+  group: sources.knative.dev
+  versions:
+    - name: v1beta1
+      served: true
+      storage: true
+      schema:
+        openAPIV3Schema:
+          type: object
+          # this is a work around so we don't need to flush out the
+          # schema for each version at this time
+          #
+          # see issue: https://github.com/knative/serving/issues/912
+          x-kubernetes-preserve-unknown-fields: true
+      subresources:
+        status: {}
+        scale:
+          # specReplicasPath defines the JSONPath inside of a custom resource that corresponds to Scale.Spec.Replicas.
+          specReplicasPath: .spec.consumers
+          # statusReplicasPath defines the JSONPath inside of a custom resource that corresponds to Scale.Status.Replicas.
+          statusReplicasPath: .status.consumers
+          # labelSelectorPath defines the JSONPath inside of a custom resource that corresponds to Scale.Status.Selector
+          labelSelectorPath: .status.selector
+      additionalPrinterColumns:
+        - name: Topics
+          type: string
+          jsonPath: ".spec.topics"
+        - name: BootstrapServers
+          type: string
+          jsonPath: ".spec.bootstrapServers"
+        - name: Ready
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].status"
+        - name: Reason
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].reason"
+        - name: Age
+          type: date
+          jsonPath: .metadata.creationTimestamp
+  names:
+    categories:
+      - all
+      - knative
+      - eventing
+      - sources
+    kind: KafkaSource
+    plural: kafkasources
+  scope: Namespaced
+  conversion:
+    strategy: Webhook
+    webhook:
+      conversionReviewVersions: ["v1", "v1beta1"]
+      clientConfig:
+        service:
+          name: kafka-source-webhook
+          namespace: knative-eventing
+
+---
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 8677448f9..515812d83 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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     app.kubernetes.io/part-of: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/high-availability: "true"
     app.kubernetes.io/name: eventing-controller
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -786,9 +786,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20211010-a046d8d24"
+        eventing.knative.dev/release: "v20211013-fcf67386c"
         app.kubernetes.io/name: eventing-autoscaler
-        app.kubernetes.io/version: "20211010-a046d8d24"
+        app.kubernetes.io/version: "20211013-fcf67386c"
         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:d266ab0b340d1f7da0e396e4df110b04c20e15f14eecb18c4996fd2df70e0ebc
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:35afd6faea97979d1eb331b141ed41b0d972513447454d638b831e32d4b3c440
           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:213f9a928827c6bf165d423c819e0f08058bb8175993da4afc69aa3e8796f176
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:0c36f336ad2e8bca705e4a7c35e9acbad37635a71bbc35491960efddb7e58037
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -868,9 +868,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     app.kubernetes.io/name: pingsource-mt-adapter
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+        eventing.knative.dev/release: "v20211013-fcf67386c"
         app.kubernetes.io/name: pingsource-mt-adapter
-        app.kubernetes.io/version: "20211010-a046d8d24"
+        app.kubernetes.io/version: "20211013-fcf67386c"
         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:3607079eddad8a294f691790c1acc1e5320ca091ad8dc24455bd4c7ea7f9fb4e
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:a7da5f04ea49804ad6a0d6ffb54a389f4efc6da231ecbc3c103cafc052927fac
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -957,9 +957,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -1027,9 +1027,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211010-a046d8d24"
+        eventing.knative.dev/release: "v20211013-fcf67386c"
         app.kubernetes.io/name: eventing-webhook
-        app.kubernetes.io/version: "20211010-a046d8d24"
+        app.kubernetes.io/version: "20211013-fcf67386c"
         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:d20e0d6d6427c5230410d75842fe5d32f5905e9ae0b06e182276af0db101bb32
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:182dfe2c54ab073fb066582869122122f2a25eedad198ae0669d935cf2e2a048
           resources:
             requests:
               # taken from serving.
@@ -1113,10 +1113,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     role: eventing-webhook
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3684,8 +3684,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4055,8 +4055,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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 d567fd2c1..c1d42515e 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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     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: "v20211010-a046d8d24"
+    eventing.knative.dev/release: "v20211013-fcf67386c"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211010-a046d8d24"
+    app.kubernetes.io/version: "20211013-fcf67386c"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/broker_prefer_reply_header_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/broker_prefer_reply_header_helper.go
new file mode 100644
index 000000000..cf52359ca
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/broker_prefer_reply_header_helper.go
@@ -0,0 +1,97 @@
+//go:build e2e
+// +build e2e
+
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package helpers
+
+import (
+	"context"
+	"testing"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/google/uuid"
+
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
+	"knative.dev/eventing/test/lib/resources"
+)
+
+func BrokerPreferHeaderCheck(
+	ctx context.Context,
+	t *testing.T,
+	creator BrokerCreator,
+	options ...testlib.SetupClientOption) {
+	const (
+		recorderName = "event-recorder"
+		triggerName  = "test-trigger"
+		senderName   = "request-sender"
+		eventSource  = "source1"
+		eventType    = "type1"
+		eventBody    = `{"msg":"test msg"}`
+	)
+
+	tests := []struct {
+		name string
+	}{
+		{
+			name: "test message without explicit prefer header should have the header",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			client := testlib.Setup(t, true, options...)
+			defer testlib.TearDown(client)
+
+			brokerName := creator(client, "v1")
+
+			// Create event tracker that should receive all events.
+			eventTracker, _ := recordevents.StartEventRecordOrFail(
+				ctx,
+				client,
+				recorderName,
+			)
+
+			client.CreateTriggerOrFail(triggerName,
+				resources.WithSubscriberServiceRefForTrigger(recorderName),
+				resources.WithBroker(brokerName),
+			)
+
+			client.WaitForAllTestResourcesReadyOrFail(ctx)
+
+			eventToSend := cloudevents.NewEvent()
+			eventToSend.SetID(uuid.New().String())
+			eventToSend.SetType(eventType)
+			eventToSend.SetSource(eventSource)
+			if err := eventToSend.SetData(cloudevents.ApplicationJSON, []byte(eventBody)); err != nil {
+				t.Fatal("Cannot set the payload of the event:", err.Error())
+			}
+			client.SendEventToAddressable(
+				ctx,
+				senderName,
+				brokerName,
+				testlib.BrokerTypeMeta,
+				eventToSend,
+			)
+
+			eventTracker.AssertAtLeast(
+				1,
+				recordevents.HasAdditionalHeader("Prefer", "reply"),
+			)
+		})
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go
index d1881157c..7776aa8ec 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go
@@ -1,18 +1,18 @@
 /*
- * Copyright 2020 The Knative Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
 
 package helpers
 
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_prefer_reply_header_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_prefer_reply_header_helper.go
new file mode 100644
index 000000000..2556267d2
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/channel_prefer_reply_header_helper.go
@@ -0,0 +1,99 @@
+//go:build e2e
+// +build e2e
+
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package helpers
+
+import (
+	"context"
+	"testing"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/google/uuid"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
+	"knative.dev/eventing/test/lib/resources"
+)
+
+func ChannelPreferHeaderCheck(
+	ctx context.Context,
+	t *testing.T,
+	component metav1.TypeMeta,
+	options ...testlib.SetupClientOption) {
+	const (
+		recorderName     = "event-recorder"
+		channelName      = "test-channel"
+		subscriptionName = "test-subscription"
+		senderName       = "request-sender"
+		eventSource      = "source1"
+		eventType        = "type1"
+		eventBody        = `{"msg":"test msg"}`
+	)
+
+	tests := []struct {
+		name string
+	}{
+		{
+			name: "test message without explicit prefer header should have it after fanout",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			client := testlib.Setup(t, true, options...)
+			defer testlib.TearDown(client)
+
+			// create channels
+			client.CreateChannelOrFail(channelName, &component)
+			client.WaitForResourcesReadyOrFail(&component)
+
+			// create loggerPod and expose it as a service
+			eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, recorderName)
+
+			client.CreateSubscriptionOrFail(
+				subscriptionName,
+				channelName,
+				&component,
+				resources.WithSubscriberForSubscription(recorderName),
+			)
+
+			client.WaitForAllTestResourcesReadyOrFail(ctx)
+
+			eventToSend := cloudevents.NewEvent()
+			eventToSend.SetID(uuid.New().String())
+			eventToSend.SetType(eventType)
+			eventToSend.SetSource(eventSource)
+			if err := eventToSend.SetData(cloudevents.ApplicationJSON, []byte(eventBody)); err != nil {
+				t.Fatal("Cannot set the payload of the event:", err.Error())
+			}
+			client.SendEventToAddressable(
+				ctx,
+				senderName,
+				channelName,
+				&component,
+				eventToSend,
+			)
+
+			eventTracker.AssertAtLeast(
+				1,
+				recordevents.HasAdditionalHeader("Prefer", "reply"),
+			)
+		})
+	}
+}