Repository: knative-sandbox/eventing-kafka-broker
Pull request #1385 (2021-10-29 07:30:42)

Add secret tracker for KafkaNetSpec
This PR adds a function `TrackNetSpecSecrets` that contains the
logic to track secrets referenced by a `KafkaNetSpec` (the type
that is embedded in `KafkaSource` and has all the secret
references)

Part of #312

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add secret tracker for KafkaNetSpec

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
None
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->
```
None
```



Patch:
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 05daa5ebf..125cecb34 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -23,7 +23,6 @@ import (
 
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/util/retry"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	"knative.dev/pkg/controller"
@@ -84,24 +83,20 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 	}
 	statusConditionManager.DataPlaneAvailable()
 
-	secret, err := security.Secret(ctx, &SecretLocator{KafkaSource: ks}, r.SecretProviderFunc())
+	authContext, err := security.ResolveAuthContextFromNetSpec(r.SecretLister, ks.GetNamespace(), ks.Spec.Net)
 	if err != nil {
-		return fmt.Errorf("failed to get secret: %w", err)
+		return fmt.Errorf("failed to create auth context: %w", err)
 	}
-	if secret != nil {
-		logger.Debug("Secret reference",
-			zap.String("apiVersion", secret.APIVersion),
-			zap.String("name", secret.Name),
-			zap.String("namespace", secret.Namespace),
-			zap.String("kind", secret.Kind),
-		)
+	secret, err := security.Secret(ctx, &SecretLocator{KafkaSource: ks}, security.NetSpecSecretProviderFunc(authContext))
+	if err != nil {
+		return fmt.Errorf("failed to get secret: %w", err)
 	}
 
 	// get security option for Sarama with secret info in it
 	securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
 
-	if err := r.TrackSecret(secret, ks); err != nil {
-		return fmt.Errorf("failed to track secret: %w", err)
+	if err := security.TrackNetSpecSecrets(r.SecretTracker, ks.Spec.Net, ks); err != nil {
+		return fmt.Errorf("failed to track secrets: %w", err)
 	}
 
 	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
@@ -141,7 +136,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
 
 	// Get resource configuration.
-	resource, err := r.reconcileKafkaSourceResource(ctx, ks, secret)
+	resource, err := r.reconcileKafkaSourceResource(ctx, ks, authContext.MultiSecretReference)
 	if err != nil {
 		return statusConditionManager.FailedToGetConfig(err)
 	}
@@ -231,8 +226,13 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *sources.KafkaSource)
 	return nil
 }
 
-func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sources.KafkaSource, secret *corev1.Secret) (*contract.Resource, error) {
-	destination, err := r.Resolver.URIFromDestinationV1(ctx, ks.Spec.Sink, ks)
+func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sources.KafkaSource, multiSecretReference *contract.MultiSecretReference) (*contract.Resource, error) {
+	destinationSpec := ks.Spec.Sink
+	if ks.Spec.Sink.Ref != nil && ks.Spec.Sink.Ref.Namespace == "" {
+		ks.Spec.Sink.DeepCopyInto(&destinationSpec)
+		destinationSpec.Ref.Namespace = ks.GetNamespace()
+	}
+	destination, err := r.Resolver.URIFromDestinationV1(ctx, destinationSpec, ks)
 	if err != nil {
 		return nil, fmt.Errorf("failed to resolve destination: %w", err)
 	}
@@ -267,15 +267,8 @@ func (r *Reconciler) reconcileKafkaSourceResource(ctx context.Context, ks *sourc
 			Extensions: ks.Spec.CloudEventOverrides.Extensions,
 		}
 	}
-	if secret != nil {
-		resource.Auth = &contract.Resource_AuthSecret{
-			AuthSecret: &contract.Reference{
-				Uuid:      string(secret.GetUID()),
-				Namespace: secret.GetNamespace(),
-				Name:      secret.GetName(),
-				Version:   secret.GetResourceVersion(),
-			},
-		}
+	if multiSecretReference != nil && len(multiSecretReference.References) > 0 {
+		resource.Auth = &contract.Resource_MultiAuthSecret{MultiAuthSecret: multiSecretReference}
 	}
 	return resource, nil
 }
diff --git a/control-plane/pkg/security/secrets_provider_net_spec.go b/control-plane/pkg/security/secrets_provider_net_spec.go
new file mode 100644
index 000000000..3bb2ce4d7
--- /dev/null
+++ b/control-plane/pkg/security/secrets_provider_net_spec.go
@@ -0,0 +1,162 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package security
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+)
+
+// NetSpecSecretProviderFunc creates a SecretProviderFunc that creates an in-memory (virtual) secret with the format
+// expected by the NewSaramaSecurityOptionFromSecret function.
+func NetSpecSecretProviderFunc(authContext *NetSpecAuthContext) SecretProviderFunc {
+	return func(_ context.Context, _, _ string) (*corev1.Secret, error) { return authContext.VirtualSecret, nil }
+}
+
+type NetSpecAuthContext struct {
+	VirtualSecret        *corev1.Secret
+	MultiSecretReference *contract.MultiSecretReference
+}
+
+// ResolveAuthContextFromNetSpec creates a NetSpecAuthContext from a provided bindings.KafkaNetSpec.
+func ResolveAuthContextFromNetSpec(lister corelisters.SecretLister, namespace string, netSpec bindings.KafkaNetSpec) (*NetSpecAuthContext, error) {
+	securityFields := []*securityField{
+		{ref: netSpec.TLS.Cert, field: contract.SecretField_USER_CRT, virtualSecretKey: UserCertificate},
+		{ref: netSpec.TLS.Key, field: contract.SecretField_USER_KEY, virtualSecretKey: UserKey},
+		{ref: netSpec.TLS.CACert, field: contract.SecretField_CA_CRT, virtualSecretKey: CaCertificateKey},
+		{ref: netSpec.SASL.Type, field: contract.SecretField_SASL_MECHANISM, virtualSecretKey: SaslMechanismKey},
+		{ref: netSpec.SASL.User, field: contract.SecretField_USER, virtualSecretKey: SaslUserKey},
+		{ref: netSpec.SASL.Password, field: contract.SecretField_PASSWORD, virtualSecretKey: SaslPasswordKey},
+	}
+	for _, b := range securityFields {
+		if err := b.resolveSecret(lister, namespace); err != nil {
+			return nil, err
+		}
+	}
+	references, virtualSecretData := toContract(securityFields)
+	multiSecretReference := &contract.MultiSecretReference{
+		Protocol:   getProtocolContractFromNetSpec(netSpec),
+		References: references,
+	}
+	virtualSecretData[ProtocolKey] = []byte(getProtocolFromNetSpec(netSpec))
+
+	authContext := &NetSpecAuthContext{
+		VirtualSecret:        &corev1.Secret{Data: virtualSecretData},
+		MultiSecretReference: multiSecretReference,
+	}
+	return authContext, nil
+}
+
+func toContract(securityFields []*securityField) ([]*contract.SecretReference, map[string][]byte) {
+	virtualSecretData := make(map[string][]byte)
+	bySecretName := make(map[string][]securityField)
+	for _, f := range securityFields {
+		if f.secret == nil || f.value == nil || len(f.value) == 0 {
+			continue
+		}
+		virtualSecretData[f.virtualSecretKey] = f.value
+		bySecretName[f.secret.Name] = append(bySecretName[f.secret.Name], *f)
+	}
+
+	refs := make([]*contract.SecretReference, 0, 6 /* max number of secrets */)
+	for secretName, securityFields := range bySecretName {
+		keyFieldReferences := make([]*contract.KeyFieldReference, 0, len(securityFields))
+		for _, f := range securityFields {
+			keyFieldReferences = append(keyFieldReferences, &contract.KeyFieldReference{
+				SecretKey: f.ref.SecretKeyRef.Key,
+				Field:     f.field,
+			})
+		}
+		any := securityFields[0]
+		refs = append(refs, &contract.SecretReference{
+			Reference: &contract.Reference{
+				Uuid:      string(any.secret.GetUID()),
+				Namespace: any.secret.GetNamespace(),
+				Name:      secretName,
+				Version:   any.secret.GetResourceVersion(),
+			},
+			KeyFieldReferences: keyFieldReferences,
+		})
+	}
+	return refs, virtualSecretData
+}
+
+type securityField struct {
+	ref              bindings.SecretValueFromSource
+	field            contract.SecretField
+	virtualSecretKey string
+
+	secret *corev1.Secret
+	value  []byte
+}
+
+func (b *securityField) resolveSecret(lister corelisters.SecretLister, namespace string) error {
+	value, secret, err := resolveSecret(lister, namespace, b.ref.SecretKeyRef)
+	if err != nil {
+		return err
+	}
+	b.value = value
+	b.secret = secret
+	return nil
+}
+
+func getProtocolFromNetSpec(netSpec bindings.KafkaNetSpec) string {
+	protocol := ProtocolPlaintext
+	if netSpec.SASL.Enable && netSpec.TLS.Enable {
+		protocol = ProtocolSASLSSL
+	} else if netSpec.SASL.Enable {
+		protocol = ProtocolSASLPlaintext
+	} else if netSpec.TLS.Enable {
+		protocol = ProtocolSSL
+	}
+	return protocol
+}
+
+func getProtocolContractFromNetSpec(netSpec bindings.KafkaNetSpec) contract.Protocol {
+	protocol := contract.Protocol_PLAINTEXT
+	if netSpec.SASL.Enable && netSpec.TLS.Enable {
+		protocol = contract.Protocol_SASL_SSL
+	} else if netSpec.SASL.Enable {
+		protocol = contract.Protocol_SASL_PLAINTEXT
+	} else if netSpec.TLS.Enable {
+		protocol = contract.Protocol_SSL
+	}
+	return protocol
+}
+
+// resolveSecret resolves the secret reference
+func resolveSecret(lister corelisters.SecretLister, ns string, ref *corev1.SecretKeySelector) ([]byte, *corev1.Secret, error) {
+	if ref == nil || ref.Name == "" {
+		return nil, nil, nil
+	}
+	secret, err := lister.Secrets(ns).Get(ref.Name)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to read secret %s/%s: %w", ns, ref.Name, err)
+	}
+
+	value, ok := secret.Data[ref.Key]
+	if !ok || len(value) == 0 {
+		return nil, nil, fmt.Errorf("missing secret key or empty secret value (%s/%s.%s)", ns, ref.Name, ref.Key)
+	}
+	return value, secret, nil
+}
diff --git a/control-plane/pkg/security/secrets_tracker.go b/control-plane/pkg/security/secrets_tracker.go
new file mode 100644
index 000000000..123b77f76
--- /dev/null
+++ b/control-plane/pkg/security/secrets_tracker.go
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package security
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	"knative.dev/pkg/tracker"
+)
+
+// TrackNetSpecSecrets tracks all secrets referenced by a provided bindings.KafkaNetSpec.
+// parent is the object that is tracking changes to those secrets.
+func TrackNetSpecSecrets(secretsTracker tracker.Interface, netSpec bindings.KafkaNetSpec, parent metav1.Object) error {
+	secrets := []bindings.SecretValueFromSource{
+		netSpec.TLS.Key,
+		netSpec.TLS.Cert,
+		netSpec.TLS.CACert,
+		netSpec.SASL.Password,
+		netSpec.SASL.User,
+		netSpec.SASL.Type,
+	}
+	for _, s := range secrets {
+		if s.SecretKeyRef != nil && s.SecretKeyRef.Name != "" {
+			ref := tracker.Reference{
+				APIVersion: "v1",
+				Kind:       "Secret",
+				Namespace:  parent.GetNamespace(),
+				Name:       s.SecretKeyRef.Name,
+			}
+			if err := secretsTracker.TrackReference(ref, parent); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesAuthProvider.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesAuthProvider.java
index f9f155db7..a8164f1d2 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesAuthProvider.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesAuthProvider.java
@@ -104,8 +104,7 @@ class KubernetesAuthProvider implements AuthProvider {
     return kubernetesClient.secrets()
       .inNamespace(secretReference.getNamespace())
       .withName(secretReference.getName())
-      .withResourceVersion(secretReference.getVersion())
-      .waitUntilReady(5, TimeUnit.SECONDS);
+      .get();
   }
 
   private static String protocolContractToSecurityProtocol(final DataPlaneContract.Protocol protocol) {
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index bccb93ae5..a131cc50f 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -26,8 +26,10 @@ import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.ext.web.client.WebClient;
+
 import java.net.URI;
 import java.util.Objects;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,6 +64,8 @@ public final class WebClientCloudEventSender implements CloudEventSender {
 
   @Override
   public Future<HttpResponse<Buffer>> send(CloudEvent event) {
+    logger.debug("Sending event {} {}", keyValue("id", event.getId()), keyValue("subscriberURI", target));
+
     TracingSpan.decorateCurrentWithEvent(event);
 
     return circuitBreaker.execute(breaker -> {
@@ -78,7 +82,10 @@ public final class WebClientCloudEventSender implements CloudEventSender {
     VertxMessageFactory
       .createWriter(client.postAbs(target))
       .writeBinary(event)
-      .onFailure(breaker::tryFail)
+      .onFailure(ex -> {
+        logError(event, ex);
+        breaker.tryFail(ex);
+      })
       .onSuccess(response -> {
 
         if (response.statusCode() >= 300 || response.statusCode() < 200) {
@@ -108,6 +115,14 @@ public final class WebClientCloudEventSender implements CloudEventSender {
     }
   }
 
+  private void logError(final CloudEvent event, Throwable ex) {
+    if (logger.isDebugEnabled()) {
+      logger.error("failed to send event to subscriber {} {}", keyValue("target", target), keyValue("event", event), ex);
+    } else {
+      logger.error("failed to send event to subscriber {}", keyValue("target", target), ex);
+    }
+  }
+
   @Override
   public Future<Void> close() {
     this.circuitBreaker.close();
diff --git a/go.mod b/go.mod
index 4b3ba7957..6fe198b8e 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.20211028061727-2eb07cbddd07
-	knative.dev/eventing-kafka v0.26.1-0.20211028065026-ee67c5013735
-	knative.dev/hack v0.0.0-20211027200727-f1228dd5e3e6
-	knative.dev/pkg v0.0.0-20211027171921-f7b70f5ce303
-	knative.dev/reconciler-test v0.0.0-20211028133527-7fae6361e997
+	knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca
+	knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c
+	knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e
+	knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40
+	knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c
 )
diff --git a/go.sum b/go.sum
index c053707cf..058e124f9 100644
--- a/go.sum
+++ b/go.sum
@@ -1306,25 +1306,27 @@ k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKU
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20211019131736-762838ee9058/go.mod h1:9mkxOcLsgMaA+LTojeXLpVoN1PW5jogk53pBXzrOQU4=
 knative.dev/eventing v0.26.1-0.20211027202426-729e39fd4218/go.mod h1:27s/uaaUrAUKO/Wy7kR4R6mbj5zugQWCyEkod4dVw0I=
-knative.dev/eventing v0.26.1-0.20211028061727-2eb07cbddd07 h1:eY5jKeaR0YdWugZJNLec3BPvuaAw/+g3lVhBBxRXVv8=
-knative.dev/eventing v0.26.1-0.20211028061727-2eb07cbddd07/go.mod h1:27s/uaaUrAUKO/Wy7kR4R6mbj5zugQWCyEkod4dVw0I=
-knative.dev/eventing-kafka v0.26.1-0.20211028065026-ee67c5013735 h1:L7Da7C4mNBc3TvpxCPHUprzjQnyTpqeoqU5aN+JrIpU=
-knative.dev/eventing-kafka v0.26.1-0.20211028065026-ee67c5013735/go.mod h1:Vz6hgwmnzvwOtOMsbqrqY8UoteyiwbRtlLV/POAdNLg=
+knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca h1:wELJavauMmbGQQ257fbAkIcZKl+AzH9fwZtD87556iA=
+knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca/go.mod h1:27s/uaaUrAUKO/Wy7kR4R6mbj5zugQWCyEkod4dVw0I=
+knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c h1:MIf1yj66j5c+eM6eHGYMjY15UosCRDsaXIVYoioGf5E=
+knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c/go.mod h1:Vz6hgwmnzvwOtOMsbqrqY8UoteyiwbRtlLV/POAdNLg=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211018110626-47ac3b032e60/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211026141922-a71c865b5f66/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211027200727-f1228dd5e3e6 h1:1mEJyeUZqScWmvHR+xHkaIRVUNPInxyNR+6lyJ/INu0=
 knative.dev/hack v0.0.0-20211027200727-f1228dd5e3e6/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e h1:0Hw2xdWYbcs2JRJnOLzAVh7APOtgro7gSno0228mnDg=
+knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20211026141922-a71c865b5f66/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20210914164111-4857ab6939e3/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
 knative.dev/pkg v0.0.0-20211018141937-a34efd6b409d/go.mod h1:/d1+Wklc8blR39M4oEBu+GeLav4vsiiJwq61tJay7mc=
 knative.dev/pkg v0.0.0-20211027105800-3b33e02e5b9c/go.mod h1:xtF+ujTgEybZxA+e7NwKgHR8xMjYzHQUBxyaOUEFxwY=
-knative.dev/pkg v0.0.0-20211027171921-f7b70f5ce303 h1:hRCMd6614KzGyGwTnfgakID3RjQb7wGn85l7ZuTOsjc=
 knative.dev/pkg v0.0.0-20211027171921-f7b70f5ce303/go.mod h1:xtF+ujTgEybZxA+e7NwKgHR8xMjYzHQUBxyaOUEFxwY=
+knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40 h1:eOOOqcj+IHIp5te9GGZsrq6vCXUv2y5JItRT3nXHs20=
+knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40/go.mod h1:HyEqMTLzT2hGCisGaRMGzEE8b4Ym+7mhmVx5ygiX+fY=
 knative.dev/reconciler-test v0.0.0-20211018075026-816dbd1c57f7/go.mod h1:6yDmb26SINSmgw6wVy9qQwgRMewiW8ddkkwGLR0ZvOY=
 knative.dev/reconciler-test v0.0.0-20211019133535-04ea7ddac7e3/go.mod h1:ltMDlEIFo5/9ES92BhIWQkdZaHRnGGJvXcYlvPxZj5s=
-knative.dev/reconciler-test v0.0.0-20211028133527-7fae6361e997 h1:HkOkCBA5CUPQpWcWpgo16eGCuK6crvDYjIU3GFD50fA=
-knative.dev/reconciler-test v0.0.0-20211028133527-7fae6361e997/go.mod h1:ltMDlEIFo5/9ES92BhIWQkdZaHRnGGJvXcYlvPxZj5s=
+knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c h1:SQ5ag46cbR0ZCJ49zc4F/+hGs0JdudsLaLpt6vSSX+E=
+knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c/go.mod h1:PIwtS0at/rtnsNCxcZUAZUmWKKasTUYJ/w1Mk94bOdA=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_scheduling.go b/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_scheduling.go
index 404b1a90b..8dfdf606d 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
@@ -18,7 +18,7 @@ package v1beta1
 
 import (
 	"k8s.io/apimachinery/pkg/types"
-	"knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing/pkg/apis/duck/v1alpha1"
 )
 
 func (k *KafkaSource) GetKey() types.NamespacedName {
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_types.go b/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_types.go
index 058095df9..7c50e50b6 100644
--- a/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_types.go
+++ b/vendor/knative.dev/eventing-kafka/pkg/apis/sources/v1beta1/kafka_types.go
@@ -19,7 +19,7 @@ package v1beta1
 import (
 	"fmt"
 
-	"knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing/pkg/apis/duck/v1alpha1"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
diff --git a/vendor/knative.dev/eventing-kafka/pkg/common/constants/constants.go b/vendor/knative.dev/eventing-kafka/pkg/common/constants/constants.go
index c6349e9e9..615679d8b 100644
--- a/vendor/knative.dev/eventing-kafka/pkg/common/constants/constants.go
+++ b/vendor/knative.dev/eventing-kafka/pkg/common/constants/constants.go
@@ -73,8 +73,4 @@ const (
 
 	// KafkaTopicConfigRetentionMs is the key in the Sarama TopicDetail ConfigEntries map for retention time (in ms)
 	KafkaTopicConfigRetentionMs = "retention.ms"
-
-	// PodAnnotationKey is an annotation used by the scheduler to be informed of pods
-	// being evicted and not use it for placing vreplicas
-	PodAnnotationKey = "eventing.knative.dev/unschedulable"
 )
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/doc.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/doc.go
similarity index 100%
rename from vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/doc.go
rename to vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/doc.go
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/placement_types.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/placement_types.go
similarity index 100%
rename from vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/placement_types.go
rename to vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/placement_types.go
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/register.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/register.go
similarity index 100%
rename from vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/register.go
rename to vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/register.go
diff --git a/vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
similarity index 100%
rename from vendor/knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
rename to vendor/knative.dev/eventing/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
index 88947290a..eeb65284e 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
@@ -25,7 +25,6 @@ import (
 )
 
 var imcCondSet = apis.NewLivingConditionSet(
-	InMemoryChannelConditionDispatcherReady,
 	InMemoryChannelConditionServiceReady,
 	InMemoryChannelConditionEndpointsReady,
 	InMemoryChannelConditionAddressable,
diff --git a/vendor/knative.dev/hack/library.sh b/vendor/knative.dev/hack/library.sh
index 42eb428e4..0877aace4 100644
--- a/vendor/knative.dev/hack/library.sh
+++ b/vendor/knative.dev/hack/library.sh
@@ -537,9 +537,11 @@ function add_trap {
 # Update go deps.
 # Parameters (parsed as flags):
 #   "--upgrade", bool, do upgrade.
-#   "--release <version>" used with upgrade. The release version to upgrade
+#   "--release <release-version>" used with upgrade. The release version to upgrade
 #                         Knative components. ex: --release v0.18. Defaults to
-#                         "master".
+#                         "main".
+#   "--module-release <module-version>" used to define a different go module tag
+#                         for a release. ex: --release v1.0 --module-release v0.27
 # Additional dependencies can be included in the upgrade by providing them in a
 # global env var: FLOATING_DEPS
 # --upgrade will set GOPROXY to direct unless it is already set.
@@ -554,13 +556,15 @@ function go_update_deps() {
   echo "=== Update Deps for Golang"
 
   local UPGRADE=0
-  local VERSION="v9000.1" # release v9000 is so far in the future, it will always pick the default branch.
+  local RELEASE="v9000.1" # release v9000 is so far in the future, it will always pick the default branch.
+  local RELEASE_MODULE=""
   local DOMAIN="knative.dev"
   while [[ $# -ne 0 ]]; do
     parameter=$1
     case ${parameter} in
       --upgrade) UPGRADE=1 ;;
-      --release) shift; VERSION="$1" ;;
+      --release) shift; RELEASE="$1" ;;
+      --module-release) shift; RELEASE_MODULE="$1" ;;
       --domain) shift; DOMAIN="$1" ;;
       *) abort "unknown option ${parameter}" ;;
     esac
@@ -568,8 +572,14 @@ function go_update_deps() {
   done
 
   if [[ $UPGRADE == 1 ]]; then
-    group "Upgrading to ${VERSION}"
-    FLOATING_DEPS+=( $(run_go_tool knative.dev/test-infra/buoy buoy float ${REPO_ROOT_DIR}/go.mod --release ${VERSION} --domain ${DOMAIN}) )
+    local buoyArgs=(--release ${RELEASE}  --domain ${DOMAIN})
+    if [ -n "$RELEASE_MODULE" ]; then
+      group "Upgrading for release ${RELEASE} to release module version ${RELEASE_MODULE}"
+      buoyArgs+=(--module-release ${RELEASE_MODULE})
+    else
+      group "Upgrading to release ${RELEASE}"
+    fi
+    FLOATING_DEPS+=( $(run_go_tool knative.dev/test-infra/buoy buoy float ${REPO_ROOT_DIR}/go.mod "${buoyArgs[@]}") )
     if [[ ${#FLOATING_DEPS[@]} > 0 ]]; then
       echo "Floating deps to ${FLOATING_DEPS[@]}"
       go get -d ${FLOATING_DEPS[@]}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 793134b39..15777f163 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1030,11 +1030,12 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.26.1-0.20211028061727-2eb07cbddd07
+# knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
 knative.dev/eventing/pkg/apis/duck/v1
+knative.dev/eventing/pkg/apis/duck/v1alpha1
 knative.dev/eventing/pkg/apis/duck/v1beta1
 knative.dev/eventing/pkg/apis/eventing
 knative.dev/eventing/pkg/apis/eventing/v1
@@ -1138,11 +1139,10 @@ 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.20211028065026-ee67c5013735
+# knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
-knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1
 knative.dev/eventing-kafka/pkg/apis/kafka
 knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1
 knative.dev/eventing-kafka/pkg/apis/messaging
@@ -1188,10 +1188,10 @@ knative.dev/eventing-kafka/pkg/common/client
 knative.dev/eventing-kafka/pkg/common/config
 knative.dev/eventing-kafka/pkg/common/constants
 knative.dev/eventing-kafka/pkg/common/kafka/sarama
-# knative.dev/hack v0.0.0-20211027200727-f1228dd5e3e6
+# knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211027171921-f7b70f5ce303
+# knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1284,7 +1284,7 @@ knative.dev/pkg/webhook/json
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20211028133527-7fae6361e997
+# knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/control-plane/pkg/security/secrets_provider_net_spec_test.go b/control-plane/pkg/security/secrets_provider_net_spec_test.go
new file mode 100644
index 000000000..85963d0ec
--- /dev/null
+++ b/control-plane/pkg/security/secrets_provider_net_spec_test.go
@@ -0,0 +1,613 @@
+/*
+ * 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 security
+
+import (
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/protobuf/testing/protocmp"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+)
+
+func TestResolveAuthContextFromNetSpec(t *testing.T) {
+	tests := []struct {
+		name                   string
+		secrets                []*corev1.Secret
+		namespace              string
+		netSpec                bindings.KafkaNetSpec
+		wantNetSpecAuthContext *NetSpecAuthContext
+		wantErr                bool
+	}{
+		{
+			name: "SASL_SSL - SCRAM-SHA-512",
+			secrets: []*corev1.Secret{
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "user"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "password"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "type"},
+					StringData: map[string]string{"key": "SCRAM-SHA-512"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cert"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "key"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cacert"},
+					StringData: map[string]string{"key": "key"},
+				},
+			},
+			namespace: "ns",
+			netSpec: bindings.KafkaNetSpec{
+				SASL: bindings.KafkaSASLSpec{
+					Enable: true,
+					User: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "user"},
+							Key:                  "key",
+						},
+					},
+					Password: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "password"},
+							Key:                  "key",
+						},
+					},
+					Type: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "type"},
+							Key:                  "key",
+						},
+					},
+				},
+				TLS: bindings.KafkaTLSSpec{
+					Enable: true,
+					Cert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cert"},
+							Key:                  "key",
+						},
+					},
+					Key: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "key"},
+							Key:                  "key",
+						},
+					},
+					CACert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cacert"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			wantNetSpecAuthContext: &NetSpecAuthContext{
+				VirtualSecret: &corev1.Secret{StringData: map[string]string{
+					ProtocolKey:      ProtocolSASLSSL,
+					CaCertificateKey: "key",
+					UserCertificate:  "key",
+					UserKey:          "key",
+					SaslMechanismKey: SaslScramSha512,
+					SaslUserKey:      "key",
+					SaslPasswordKey:  "key",
+				}},
+				MultiSecretReference: &contract.MultiSecretReference{
+					Protocol: contract.Protocol_SASL_SSL,
+					References: []*contract.SecretReference{
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "cert"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_CRT},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "key"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_KEY},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "cacert"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_CA_CRT},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "type"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_SASL_MECHANISM},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "user"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "password"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_PASSWORD},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "SASL_SSL - SCRAM-SHA-256",
+			secrets: []*corev1.Secret{
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "user"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "password"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "type"},
+					StringData: map[string]string{"key": "SCRAM-SHA-256"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cert"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "key"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cacert"},
+					StringData: map[string]string{"key": "key"},
+				},
+			},
+			namespace: "ns",
+			netSpec: bindings.KafkaNetSpec{
+				SASL: bindings.KafkaSASLSpec{
+					Enable: true,
+					User: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "user"},
+							Key:                  "key",
+						},
+					},
+					Password: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "password"},
+							Key:                  "key",
+						},
+					},
+					Type: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "type"},
+							Key:                  "key",
+						},
+					},
+				},
+				TLS: bindings.KafkaTLSSpec{
+					Enable: true,
+					Cert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cert"},
+							Key:                  "key",
+						},
+					},
+					Key: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "key"},
+							Key:                  "key",
+						},
+					},
+					CACert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cacert"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			wantNetSpecAuthContext: &NetSpecAuthContext{
+				VirtualSecret: &corev1.Secret{StringData: map[string]string{
+					ProtocolKey:      ProtocolSASLSSL,
+					CaCertificateKey: "key",
+					UserCertificate:  "key",
+					UserKey:          "key",
+					SaslMechanismKey: SaslScramSha256,
+					SaslUserKey:      "key",
+					SaslPasswordKey:  "key",
+				}},
+				MultiSecretReference: &contract.MultiSecretReference{
+					Protocol: contract.Protocol_SASL_SSL,
+					References: []*contract.SecretReference{
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "cert"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_CRT},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "key"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_KEY},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "cacert"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_CA_CRT},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "type"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_SASL_MECHANISM},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "user"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "password"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_PASSWORD},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "SSL",
+			secrets: []*corev1.Secret{
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cert"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "key"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cacert"},
+					StringData: map[string]string{"key": "key"},
+				},
+			},
+			namespace: "ns",
+			netSpec: bindings.KafkaNetSpec{
+				TLS: bindings.KafkaTLSSpec{
+					Enable: true,
+					Cert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cert"},
+							Key:                  "key",
+						},
+					},
+					Key: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "key"},
+							Key:                  "key",
+						},
+					},
+					CACert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cacert"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			wantNetSpecAuthContext: &NetSpecAuthContext{
+				VirtualSecret: &corev1.Secret{StringData: map[string]string{
+					ProtocolKey:      ProtocolSSL,
+					CaCertificateKey: "key",
+					UserCertificate:  "key",
+					UserKey:          "key",
+				}},
+				MultiSecretReference: &contract.MultiSecretReference{
+					Protocol: contract.Protocol_SSL,
+					References: []*contract.SecretReference{
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "cert"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_CRT},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "key"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_KEY},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "cacert"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_CA_CRT},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "SSL - no CA",
+			secrets: []*corev1.Secret{
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cert"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "key"},
+					StringData: map[string]string{"key": "key"},
+				},
+			},
+			namespace: "ns",
+			netSpec: bindings.KafkaNetSpec{
+				TLS: bindings.KafkaTLSSpec{
+					Enable: true,
+					Cert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cert"},
+							Key:                  "key",
+						},
+					},
+					Key: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "key"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			wantNetSpecAuthContext: &NetSpecAuthContext{
+				VirtualSecret: &corev1.Secret{StringData: map[string]string{
+					ProtocolKey:     ProtocolSSL,
+					UserCertificate: "key",
+					UserKey:         "key",
+				}},
+				MultiSecretReference: &contract.MultiSecretReference{
+					Protocol: contract.Protocol_SSL,
+					References: []*contract.SecretReference{
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "cert"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_CRT},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "key"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER_KEY},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "SASL_PLAINTEXT - SCRAM-SHA-256",
+			secrets: []*corev1.Secret{
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "user"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "password"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "type"},
+					StringData: map[string]string{"key": "SCRAM-SHA-256"},
+				},
+			},
+			namespace: "ns",
+			netSpec: bindings.KafkaNetSpec{
+				SASL: bindings.KafkaSASLSpec{
+					Enable: true,
+					User: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "user"},
+							Key:                  "key",
+						},
+					},
+					Password: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "password"},
+							Key:                  "key",
+						},
+					},
+					Type: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "type"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			wantNetSpecAuthContext: &NetSpecAuthContext{
+				VirtualSecret: &corev1.Secret{StringData: map[string]string{
+					ProtocolKey:      ProtocolSASLPlaintext,
+					SaslMechanismKey: SaslScramSha256,
+					SaslUserKey:      "key",
+					SaslPasswordKey:  "key",
+				}},
+				MultiSecretReference: &contract.MultiSecretReference{
+					Protocol: contract.Protocol_SASL_PLAINTEXT,
+					References: []*contract.SecretReference{
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "type"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_SASL_MECHANISM},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "user"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "password"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_PASSWORD},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "SASL_PLAINTEXT - SCRAM-SHA-512",
+			secrets: []*corev1.Secret{
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "user"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "password"},
+					StringData: map[string]string{"key": "key"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "type"},
+					StringData: map[string]string{"key": "SCRAM-SHA-512"},
+				},
+			},
+			namespace: "ns",
+			netSpec: bindings.KafkaNetSpec{
+				SASL: bindings.KafkaSASLSpec{
+					Enable: true,
+					User: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "user"},
+							Key:                  "key",
+						},
+					},
+					Password: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "password"},
+							Key:                  "key",
+						},
+					},
+					Type: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "type"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			wantNetSpecAuthContext: &NetSpecAuthContext{
+				VirtualSecret: &corev1.Secret{StringData: map[string]string{
+					ProtocolKey:      ProtocolSASLPlaintext,
+					SaslMechanismKey: SaslScramSha512,
+					SaslUserKey:      "key",
+					SaslPasswordKey:  "key",
+				}},
+				MultiSecretReference: &contract.MultiSecretReference{
+					Protocol: contract.Protocol_SASL_PLAINTEXT,
+					References: []*contract.SecretReference{
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "type"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_SASL_MECHANISM},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "user"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_USER},
+							},
+						},
+						{
+							Reference: &contract.Reference{Namespace: "ns", Name: "password"},
+							KeyFieldReferences: []*contract.KeyFieldReference{
+								{SecretKey: "key", Field: contract.SecretField_PASSWORD},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, _ := reconcilertesting.SetupFakeContext(t)
+			lister := secretinformer.Get(ctx)
+			for _, s := range tt.secrets {
+				cp := copySecretDataToData(s)
+				_ = lister.Informer().GetStore().Add(cp)
+			}
+
+			got, err := ResolveAuthContextFromNetSpec(lister.Lister(), tt.namespace, tt.netSpec)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("error = %v, wantErr %v", err, tt.wantErr)
+			}
+			wantSecret := copySecretDataToData(tt.wantNetSpecAuthContext.VirtualSecret)
+			wantSecret.StringData = nil
+			tt.wantNetSpecAuthContext.VirtualSecret = wantSecret
+
+			if diff := cmp.Diff(tt.wantNetSpecAuthContext.VirtualSecret, got.VirtualSecret); diff != "" {
+				t.Error("(-want, +got)", diff)
+			}
+			wantMS := tt.wantNetSpecAuthContext.MultiSecretReference
+			sort.Slice(wantMS.References, func(i, j int) bool {
+				return strings.Compare(wantMS.References[i].Reference.Name, wantMS.References[j].Reference.Name) < 0
+			})
+			gotMS := got.MultiSecretReference
+			sort.Slice(gotMS.References, func(i, j int) bool {
+				return strings.Compare(gotMS.References[i].Reference.Name, gotMS.References[j].Reference.Name) < 0
+			})
+			if diff := cmp.Diff(wantMS, gotMS, protocmp.Transform()); diff != "" {
+				t.Error("(-want, +got)", diff)
+			}
+		})
+	}
+}
+
+func copySecretDataToData(s *corev1.Secret) *corev1.Secret {
+	cp := s.DeepCopy()
+	if cp.Data == nil {
+		cp.Data = map[string][]byte{}
+	}
+	for k, v := range cp.StringData {
+		cp.Data[k] = []byte(v)
+	}
+	return cp
+}
diff --git a/control-plane/pkg/security/secrets_tracker_test.go b/control-plane/pkg/security/secrets_tracker_test.go
new file mode 100644
index 000000000..c990ac719
--- /dev/null
+++ b/control-plane/pkg/security/secrets_tracker_test.go
@@ -0,0 +1,168 @@
+/*
+ * 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 security
+
+import (
+	"io"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/pkg/tracker"
+)
+
+func TestTrackNetSpecSecrets(t *testing.T) {
+	tests := []struct {
+		name                        string
+		secretsTracker              *mockTracker
+		netSpec                     bindings.KafkaNetSpec
+		expectedTrackReferenceCalls int
+		parent                      metav1.Object
+		wantErr                     bool
+	}{
+		{
+			name:           "track all",
+			secretsTracker: &mockTracker{},
+			parent: &sources.KafkaSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "s",
+					Namespace: "ns",
+				},
+			},
+			netSpec: bindings.KafkaNetSpec{
+				SASL: bindings.KafkaSASLSpec{
+					Enable: true,
+					User: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "user"},
+							Key:                  "key",
+						},
+					},
+					Password: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "password"},
+							Key:                  "key",
+						},
+					},
+					Type: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "type"},
+							Key:                  "key",
+						},
+					},
+				},
+				TLS: bindings.KafkaTLSSpec{
+					Enable: true,
+					Cert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cert"},
+							Key:                  "key",
+						},
+					},
+					Key: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "key"},
+							Key:                  "key",
+						},
+					},
+					CACert: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "cacert"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			expectedTrackReferenceCalls: 6,
+			wantErr:                     false,
+		},
+		{
+			name:           "track error",
+			secretsTracker: &mockTracker{trackReferenceErr: io.EOF},
+			parent: &sources.KafkaSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "s",
+					Namespace: "ns",
+				},
+			},
+			netSpec: bindings.KafkaNetSpec{
+				SASL: bindings.KafkaSASLSpec{
+					Enable: true,
+					User: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "user"},
+							Key:                  "key",
+						},
+					},
+					Password: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "password"},
+							Key:                  "key",
+						},
+					},
+					Type: bindings.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{Name: "type"},
+							Key:                  "key",
+						},
+					},
+				},
+			},
+			expectedTrackReferenceCalls: 1,
+			wantErr:                     true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := TrackNetSpecSecrets(tt.secretsTracker, tt.netSpec, tt.parent); (err != nil) != tt.wantErr {
+				t.Errorf("TrackNetSpecSecrets() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if tt.secretsTracker.trackReferenceCalls != tt.expectedTrackReferenceCalls {
+				t.Errorf("Expected %d calls to TrackReference, got %d", tt.expectedTrackReferenceCalls, tt.secretsTracker.trackReferenceCalls)
+			}
+		})
+	}
+}
+
+type mockTracker struct {
+	trackReferenceCalls int
+	trackReferenceErr   error
+}
+
+func (m mockTracker) Track(corev1.ObjectReference, interface{}) error {
+	panic("implement me")
+}
+
+func (m *mockTracker) TrackReference(tracker.Reference, interface{}) error {
+	m.trackReferenceCalls++
+	return m.trackReferenceErr
+}
+
+func (m mockTracker) OnChanged(interface{}) {
+	panic("implement me")
+}
+
+func (m mockTracker) GetObservers(interface{}) []types.NamespacedName {
+	panic("implement me")
+}
+
+func (m mockTracker) OnDeletedObserver(interface{}) {
+	panic("implement me")
+}
diff --git a/test/e2e_new/broker_not_ready_test.go b/test/e2e_new/broker_not_ready_test.go
new file mode 100644
index 000000000..51d3f8b20
--- /dev/null
+++ b/test/e2e_new/broker_not_ready_test.go
@@ -0,0 +1,81 @@
+// +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 e2e_new
+
+import (
+	"testing"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	brokerconfigmap "knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap/broker"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+func TestBrokerNotReadyAfterBeingReady(t *testing.T) {
+
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, BrokerNotReadyAfterBeingReady())
+}
+
+func BrokerNotReadyAfterBeingReady() *feature.Feature {
+	f := feature.NewFeature()
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	configName := feature.MakeRandomK8sName("config")
+
+	f.Setup("create broker config", brokerconfigmap.Install(
+		configName,
+		brokerconfigmap.WithBootstrapServer("my-cluster-kafka-bootstrap.kafka:9092"),
+		brokerconfigmap.WithNumPartitions(1),
+		brokerconfigmap.WithReplicationFactor(1),
+	))
+
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithConfig(configName),
+	))
+
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+
+	f.Requirement("update broker config", brokerconfigmap.Install(
+		configName,
+		// Invalid bootstrap server
+		brokerconfigmap.WithBootstrapServer("my-cluster-kafka-bootstrap.non-kafka:9092"),
+		brokerconfigmap.WithNumPartitions(1),
+		brokerconfigmap.WithReplicationFactor(1),
+	))
+
+	f.Assert("broker is not ready", k8s.IsNotReady(broker.GVR(), brokerName))
+
+	return f
+}
diff --git a/test/e2e_new/resources/configmap/broker/broker_config.go b/test/e2e_new/resources/configmap/broker/broker_config.go
new file mode 100644
index 000000000..6b0df2495
--- /dev/null
+++ b/test/e2e_new/resources/configmap/broker/broker_config.go
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     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 broker
+
+import (
+	"context"
+	"embed"
+
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name": name,
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+// WithReplicationFactor adds the replication factor
+func WithReplicationFactor(replicationFactor int16) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["replicationFactor"] = replicationFactor
+	}
+}
+
+// WithNumPartitions adds the num of partitions
+func WithNumPartitions(numPartitions int32) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["numPartitions"] = numPartitions
+	}
+}
+
+// WithBootstrapServer adds the bootstrap server
+func WithBootstrapServer(bootstrapServer string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["bootstrapServer"] = bootstrapServer
+	}
+}
diff --git a/test/e2e_new/resources/configmap/broker/broker_config.yaml b/test/e2e_new/resources/configmap/broker/broker_config.yaml
new file mode 100644
index 000000000..7648bd7ce
--- /dev/null
+++ b/test/e2e_new/resources/configmap/broker/broker_config.yaml
@@ -0,0 +1,29 @@
+# Copyright 2021 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+data:
+  {{ if .bootstrapServer }}
+  bootstrap.servers: "{{ .bootstrapServer }}"
+  {{ end }}
+  {{ if .numPartitions }}
+  default.topic.partitions: "{{ .numPartitions }}"
+  {{ end }}
+  {{ if .replicationFactor }}
+  default.topic.replication.factor: "{{ .replicationFactor }}"
+  {{ end }}
diff --git a/test/kafka/kafka-ephemeral.yaml b/test/kafka/kafka-ephemeral.yaml
index ff8f35f8d..1ef56903b 100644
--- a/test/kafka/kafka-ephemeral.yaml
+++ b/test/kafka/kafka-ephemeral.yaml
@@ -33,20 +33,20 @@ spec:
         tls: true
         authentication:
           type: tls
-      # protocol=SASL_PLAINTEXT
+      # protocol=SASL_SSL
       # sasl.mechanism=SCRAM-SHA-512
-      - name: saslplain
+      - name: saslssl
         port: 9094
         type: internal
-        tls: false
+        tls: true
         authentication:
           type: scram-sha-512
-      # protocol=SASL_SSL
+      # protocol=SASL_PLAINTEXT
       # sasl.mechanism=SCRAM-SHA-512
-      - name: saslssl
+      - name: saslplain
         port: 9095
         type: internal
-        tls: true
+        tls: false
         authentication:
           type: scram-sha-512
     config:
diff --git a/test/kafka/kafka_setup.sh b/test/kafka/kafka_setup.sh
index aa9f44f85..60209f992 100755
--- a/test/kafka/kafka_setup.sh
+++ b/test/kafka/kafka_setup.sh
@@ -14,13 +14,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-set -e
+set -euo pipefail
 
 source $(dirname $0)/../../vendor/knative.dev/hack/e2e-tests.sh
 
-kubectl create namespace kafka --dry-run -o yaml | kubectl apply -f -
+CLUSTER_SUFFIX=${CLUSTER_SUFFIX:-"cluster.local"}
+SYSTEM_NAMESPACE=${SYSTEM_NAMESPACE:-"knative-eventing"}
 
-sleep 5
+kubectl create namespace kafka --dry-run -o yaml | kubectl apply -f -
 
 header "Applying Strimzi Cluster Operator file"
 cat $(dirname $0)/strimzi-cluster-operator.yaml | sed "s/cluster.local/${CLUSTER_SUFFIX}/g" | kubectl apply -n kafka -f -
@@ -35,6 +36,75 @@ kubectl apply -n kafka -f $(dirname $0)/user-tls.yaml
 echo "Create SASL SCRUM 512 user"
 kubectl apply -n kafka -f $(dirname $0)/user-sasl-scram-512.yaml
 
-sleep 5
+# Waits until the given object exists.
+# Parameters: $1 - the kind of the object.
+#             $2 - object's name.
+#             $3 - namespace (optional).
+function wait_until_object_exists() {
+  local KUBECTL_ARGS="get $1 $2"
+  local DESCRIPTION="$1 $2"
+
+  if [[ -n $3 ]]; then
+    KUBECTL_ARGS="get -n $3 $1 $2"
+    DESCRIPTION="$1 $3/$2"
+  fi
+  echo -n "Waiting until ${DESCRIPTION} exists"
+  for i in {1..150}; do  # timeout after 5 minutes
+    if kubectl ${KUBECTL_ARGS} > /dev/null 2>&1; then
+      echo -e "\n${DESCRIPTION} exists"
+      return 0
+    fi
+    echo -n "."
+    sleep 2
+  done
+  echo -e "\n\nERROR: timeout waiting for ${DESCRIPTION} to exist"
+  kubectl ${KUBECTL_ARGS}
+  return 1
+}
+
+function create_tls_secrets() {
+  ca_cert_secret="my-cluster-cluster-ca-cert"
+  tls_user="my-tls-user"
+
+  echo "Waiting until secrets: ${ca_cert_secret}, ${tls_user} don't exist"
+  wait_until_object_exists secret ${ca_cert_secret} kafka
+  wait_until_object_exists secret ${tls_user} kafka
+
+  echo "Creating TLS Kafka secret"
+
+  STRIMZI_CRT=$(kubectl -n kafka get secret "${ca_cert_secret}" --template='{{index .data "ca.crt"}}' | base64 --decode )
+  TLSUSER_CRT=$(kubectl -n kafka get secret "${tls_user}" --template='{{index .data "user.crt"}}' | base64 --decode )
+  TLSUSER_KEY=$(kubectl -n kafka get secret "${tls_user}" --template='{{index .data "user.key"}}' | base64 --decode )
+
+  kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-tls-secret \
+    --from-literal=ca.crt="$STRIMZI_CRT" \
+    --from-literal=user.crt="$TLSUSER_CRT" \
+    --from-literal=user.key="$TLSUSER_KEY" \
+    --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
+}
+
+function create_sasl_secrets() {
+  ca_cert_secret="my-cluster-cluster-ca-cert"
+  sasl_user="my-sasl-user"
+
+  echo "Waiting until secrets: ${ca_cert_secret}, ${sasl_user} don't exist"
+  wait_until_object_exists secret ${ca_cert_secret} kafka
+  wait_until_object_exists secret ${sasl_user} kafka
+
+  echo "Creating SASL Kafka secret"
+
+  STRIMZI_CRT=$(kubectl -n kafka get secret ${ca_cert_secret} --template='{{index .data "ca.crt"}}' | base64 --decode )
+  SASL_PASSWD=$(kubectl -n kafka get secret ${sasl_user} --template='{{index .data "password"}}' | base64 --decode )
+
+  kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-secret \
+    --from-literal=ca.crt="$STRIMZI_CRT" \
+    --from-literal=password="$SASL_PASSWD" \
+    --from-literal=saslType="SCRAM-SHA-512" \
+    --from-literal=user="my-sasl-user" \
+    --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
+}
 
 wait_until_pods_running kafka || fail_test "Failed to start up a Kafka cluster"
+
+create_sasl_secrets || fail_test "Failed to create SASL secrets"
+create_tls_secrets || fail_test "Failed to create TLS secrets"
diff --git a/test/pkg/testing/run.go b/test/pkg/testing/run.go
index e21cb807a..f938e3b81 100644
--- a/test/pkg/testing/run.go
+++ b/test/pkg/testing/run.go
@@ -28,8 +28,8 @@ const (
 	// Kafka bootstrap server.
 	BootstrapServersPlaintext     = "my-cluster-kafka-bootstrap.kafka:9092"
 	BootstrapServersSsl           = "my-cluster-kafka-bootstrap.kafka:9093"
-	BootstrapServersSaslPlaintext = "my-cluster-kafka-bootstrap.kafka:9094"
-	BootstrapServersSslSaslScram  = "my-cluster-kafka-bootstrap.kafka:9095"
+	BootstrapServersSaslPlaintext = "my-cluster-kafka-bootstrap.kafka:9095"
+	BootstrapServersSslSaslScram  = "my-cluster-kafka-bootstrap.kafka:9094"
 
 	KafkaClusterNamespace = "kafka"
 	TlsUserSecretName     = "my-tls-user"
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 98128c4ae..23ac2842d 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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -472,7 +472,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -521,8 +521,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "96896b00"
@@ -583,10 +583,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -637,10 +637,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -714,10 +714,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "4002b4c2"
@@ -777,10 +777,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/high-availability: "true"
     app.kubernetes.io/name: eventing-controller
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -790,9 +790,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20211026-a20ff702a"
+        eventing.knative.dev/release: "v20211028-2eb07cbdd"
         app.kubernetes.io/name: eventing-autoscaler
-        app.kubernetes.io/version: "20211026-a20ff702a"
+        app.kubernetes.io/version: "20211028-2eb07cbdd"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -810,7 +810,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:496ba4b1248ae332ff67c73b7edfab2e5583c3a13836654461c1985420d03f51
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:570deacf9cb8c9768c800fc7853abd56f8a3aa478d804223224ddd462c1662c2
           resources:
             requests:
               cpu: 100m
@@ -828,7 +828,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:a83e3b6264accfbee34a03ea66e49855b66973f798971ffbbd1b4c8483767f9c
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:328aefcff1d37d31e3d113291fc38be0e3266d9e4158e62956bab8982bafaf12
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -872,9 +872,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     app.kubernetes.io/name: pingsource-mt-adapter
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
@@ -887,9 +887,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211026-a20ff702a"
+        eventing.knative.dev/release: "v20211028-2eb07cbdd"
         app.kubernetes.io/name: pingsource-mt-adapter
-        app.kubernetes.io/version: "20211026-a20ff702a"
+        app.kubernetes.io/version: "20211028-2eb07cbdd"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       affinity:
@@ -903,7 +903,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:d97581f294000f6faeb2be292a864d2be2a785d4bb866189f391d7e6326e7226
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:486742ebf0fb7cf26cc50be336351c583d546f9a52195bcd297d5d6f5b74d71d
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -961,9 +961,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   scaleTargetRef:
@@ -987,9 +987,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   minAvailable: 80%
@@ -1018,9 +1018,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -1031,9 +1031,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211026-a20ff702a"
+        eventing.knative.dev/release: "v20211028-2eb07cbdd"
         app.kubernetes.io/name: eventing-webhook
-        app.kubernetes.io/version: "20211026-a20ff702a"
+        app.kubernetes.io/version: "20211028-2eb07cbdd"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1053,7 +1053,7 @@ spec:
           terminationMessagePolicy: FallbackToLogsOnError
           # This is the Go import path for the binary that is containerized
           # and substituted here.
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:5cb8e83d324efdacc06fca6d3f11a18d0b5f330155292b1fc72c3a65c47018b2
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:301b6c69691ec28d75f228f2b88d9bea8b875dd6081c6a82fdbe8869c13c4300
           resources:
             requests:
               # taken from serving.
@@ -1117,10 +1117,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     role: eventing-webhook
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1152,11 +1152,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1368,10 +1368,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1535,11 +1535,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1807,11 +1807,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1961,9 +1961,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2090,10 +2090,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2345,11 +2345,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2536,10 +2536,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2857,12 +2857,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3049,9 +3049,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3261,9 +3261,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3446,8 +3446,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3460,9 +3460,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3480,9 +3480,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3503,9 +3503,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3530,9 +3530,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3548,35 +3548,12 @@ rules:
 ---
 kind: ClusterRole
 apiVersion: rbac.authorization.k8s.io/v1
-metadata:
-  name: messaging-addressable-resolver
-  labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
-    app.kubernetes.io/part-of: knative-eventing
-# Do not use this role directly. These rules will be added to the "addressable-resolver" role.
-rules:
-  - apiGroups:
-      - messaging.knative.dev
-    resources:
-      - sequences
-      - sequences/status
-      - parallels
-      - parallels/status
-    verbs:
-      - get
-      - list
-      - watch
----
-kind: ClusterRole
-apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3612,8 +3589,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3639,8 +3616,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3657,8 +3634,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3691,8 +3668,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3705,9 +3682,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3745,9 +3722,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3759,9 +3736,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3773,9 +3750,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3787,9 +3764,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3801,9 +3778,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3816,8 +3793,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3830,8 +3807,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3858,8 +3835,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3994,8 +3971,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4062,8 +4039,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4076,9 +4053,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4124,8 +4101,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4138,9 +4115,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4176,8 +4153,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4263,8 +4240,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4363,8 +4340,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4400,8 +4377,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4438,8 +4415,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4472,8 +4449,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4507,8 +4484,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 # The data is populated at install time.
 
@@ -4532,8 +4509,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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 3e1aa5695..3f6f01485 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: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     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: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -400,11 +400,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -672,11 +672,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -826,9 +826,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -955,10 +955,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1210,11 +1210,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1401,10 +1401,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1722,12 +1722,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1914,9 +1914,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2126,9 +2126,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211026-a20ff702a"
+    eventing.knative.dev/release: "v20211028-2eb07cbdd"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211026-a20ff702a"
+    app.kubernetes.io/version: "20211028-2eb07cbdd"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev