Repository: knative-sandbox/eventing-kafka-broker
Pull request #670 (2021-02-21 22:04:33)

Add KafkaSink docs link to the readme
## Proposed Changes

- Add KafkaSink docs link to the readme

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **pierdipi** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @pierdipi` in a comment when ready.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["pierdipi"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670?src=pr&el=h1) Report
> Merging [#670](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670?src=pr&el=desc) (2de7350) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/182da5c2929cc745f5f62d66d0eca18d67ce0278?el=desc) (182da5c) will **decrease** coverage by `9.90%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670?src=pr&el=tree)

```diff
@@             Coverage Diff              @@
##             master     #670      +/-   ##
============================================
- Coverage     73.44%   63.53%   -9.91%     
============================================
  Files            77       28      -49     
  Lines          2851     1215    -1636     
  Branches        128        0     -128     
============================================
- Hits           2094      772    -1322     
+ Misses          611      366     -245     
+ Partials        146       77      -69     
```

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

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...patcher/consumer/impl/OrderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvY29uc3VtZXIvaW1wbC9PcmRlcmVkQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | | | |
| [...knative/eventing/kafka/broker/dispatcher/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvTWFpbi5qYXZh) | | | |
| [...fka/broker/core/security/PlaintextCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvUGxhaW50ZXh0Q3JlZGVudGlhbHMuamF2YQ==) | | | |
| [...tive/eventing/kafka/broker/core/filter/Filter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsdGVyL0ZpbHRlci5qYXZh) | | | |
| [...oker/dispatcher/http/HttpConsumerRecordSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaHR0cC9IdHRwQ29uc3VtZXJSZWNvcmRTZW5kZXIuamF2YQ==) | | | |
| [...kafka/broker/core/tracing/OpenTelemetryTracer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9PcGVuVGVsZW1ldHJ5VHJhY2VyLmphdmE=) | | | |
| [...ka/broker/dispatcher/ConsumerDeployerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvQ29uc3VtZXJEZXBsb3llclZlcnRpY2xlLmphdmE=) | | | |
| [...ka/broker/core/security/KubernetesCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS3ViZXJuZXRlc0NyZWRlbnRpYWxzLmphdmE=) | | | |
| [...a/broker/core/tracing/HeadersPropagatorSetter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9IZWFkZXJzUHJvcGFnYXRvclNldHRlci5qYXZh) | | | |
| [...afka/broker/core/filter/impl/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsdGVyL2ltcGwvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | | |
| ... and [39 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670?src=pr&el=footer). Last update [182da5c...2de7350](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/670?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

pierDipi:  /assign @slinkydeveloper

Patch:
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
new file mode 100644
index 000000000..99b03cfbe
--- /dev/null
+++ b/.github/CODEOWNERS
@@ -0,0 +1,9 @@
+# Default owners are TOC, release leads and Eventing writers
+* @knative-sandbox/technical-oversight-committee
+* @knative-sandbox/knative-release-leads
+* @knative-sandbox/eventing-writers
+* @knative-sandbox/eventing-kafka-broker-approvers
+
+# hack and test are owned by productivity
+/hack/* @knative-sandbox/productivity-wg-leads
+/test/* @knative-sandbox/productivity-wg-leads
diff --git a/README.md b/README.md
index 15ac32f88..b4ff1b76a 100644
--- a/README.md
+++ b/README.md
@@ -36,6 +36,7 @@
 
 <p align="center">
   <a href="https://knative.dev/docs/eventing/broker/kafka-broker/">Eventing Kafka Broker Docs</a> •
+  <a href="https://knative.dev/docs/eventing/sink/kafka-sink/">Eventing Kafka Sink Docs</a> •
   <a href="https://github.com/knative-sandbox/eventing-kafka-broker/blob/master/CONTRIBUTING.md">Contributing</a> •
   <a href="https://github.com/issues?q=is%3Aopen+is%3Aissue+archived%3Afalse+org%3Aknative-sandbox+repo%3Aeventing-kafka-broker+label%3A%22help+wanted%22+%22+">Help Wanted Issues</a>
 </p>
diff --git a/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/controller.go b/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/controller.go
index af6740b4a..2625c3f7e 100644
--- a/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/controller.go
+++ b/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/reconciler.go b/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/reconciler.go
index 95ca97488..a7ea654dd 100644
--- a/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/reconciler.go
+++ b/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/control-plane/pkg/security/secret.go b/control-plane/pkg/security/secret.go
index efc78ed2e..822c26d89 100644
--- a/control-plane/pkg/security/secret.go
+++ b/control-plane/pkg/security/secret.go
@@ -20,6 +20,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	"strconv"
 
 	"github.com/Shopify/sarama"
 )
@@ -31,6 +32,7 @@ const (
 
 	UserCertificate = "user.crt"
 	UserKey         = "user.key"
+	UserSkip        = "user.skip" // default: false
 
 	SaslMechanismKey = "sasl.mechanism"
 	SaslUserKey      = "user"
@@ -149,32 +151,54 @@ func saslConfig(protocol string, data map[string][]byte) ConfigOption {
 func sslConfig(protocol string, data map[string][]byte) ConfigOption {
 	return func(config *sarama.Config) error {
 
-		caCert, ok := data[CaCertificateKey]
-		if !ok {
-			return fmt.Errorf("[protocol %s] required CA certificate (key: %s)", protocol, CaCertificateKey)
+		// If there is no CaCertificateKey, we assume that we can use system's root CA set.
+		var certPool *x509.CertPool
+		if caCert, ok := data[CaCertificateKey]; ok {
+			certPool = x509.NewCertPool()
+			if !certPool.AppendCertsFromPEM(caCert) {
+				return fmt.Errorf(
+					"[protocol %s] failed to parse CA certificates (key: %s)",
+					protocol, CaCertificateKey,
+				)
+			}
 		}
 
 		var tlsCerts []tls.Certificate
 		if protocol == ProtocolSSL {
-			userKeyCert, ok := data[UserKey]
-			if !ok {
-				return fmt.Errorf("[protocol %s] required user key (key: %s)", protocol, UserKey)
-			}
 
-			userCert, ok := data[UserCertificate]
-			if !ok {
-				return fmt.Errorf("[protocol %s] required user key (key: %s)", protocol, UserCertificate)
-			}
+			// When protocol=SSL we might have 2 options:
+			//	- client auth is required on the broker side (mTLS)
+			//	- client auth is not required on the broker side
 
-			tlsCert, err := tls.X509KeyPair(userCert, userKeyCert)
+			skipClientAuth, err := skipClientAuthCheck(data)
 			if err != nil {
-				return fmt.Errorf("[protocol %s] failed to load x.509 key pair: %w", protocol, err)
+				return fmt.Errorf("[protocol %s] %w", protocol, err)
 			}
-			tlsCerts = []tls.Certificate{tlsCert}
-		}
 
-		certPool := x509.NewCertPool()
-		certPool.AppendCertsFromPEM(caCert)
+			if !skipClientAuth {
+				userKeyCert, ok := data[UserKey]
+				if !ok {
+					return fmt.Errorf(
+						`[protocol %s] required user key (key: %s) - use "%s: true" to disable client auth`,
+						protocol, UserKey, UserSkip,
+					)
+				}
+
+				userCert, ok := data[UserCertificate]
+				if !ok {
+					return fmt.Errorf(
+						`[protocol %s] required user key (key: %s) - use "%s: true" to disable client auth`,
+						protocol, UserCertificate, UserSkip,
+					)
+				}
+
+				tlsCert, err := tls.X509KeyPair(userCert, userKeyCert)
+				if err != nil {
+					return fmt.Errorf("[protocol %s] failed to load x.509 key pair: %w", protocol, err)
+				}
+				tlsCerts = []tls.Certificate{tlsCert}
+			}
+		}
 
 		config.Net.TLS.Enable = true
 		config.Net.TLS.Config = &tls.Config{
@@ -187,3 +211,15 @@ func sslConfig(protocol string, data map[string][]byte) ConfigOption {
 		return nil
 	}
 }
+
+func skipClientAuthCheck(data map[string][]byte) (bool, error) {
+	clientAuth, ok := data[UserSkip]
+	if !ok {
+		return false, nil // Client auth is required by default.
+	}
+	enabled, err := strconv.ParseBool(string(clientAuth))
+	if err != nil {
+		return false, fmt.Errorf("failed to parse client auth flag (key: %s): %w", UserSkip, err)
+	}
+	return enabled, nil
+}
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java
index 480615c1f..6879a4f4f 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java
@@ -25,6 +25,13 @@ public interface Credentials {
    */
   String caCertificates();
 
+  /**
+   * Skip client auth.
+   *
+   * @return true if client auth should be skipped otherwise false
+   */
+  boolean skipClientAuth();
+
   /**
    * @return user certificate.
    */
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
index d0e3aebc2..26bbbaec5 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
@@ -35,7 +35,10 @@ class CredentialsValidator {
     }
 
     if (is(SecurityProtocol.SSL, securityProtocol)) {
-      if (anyBlank(credentials.userCertificate(), credentials.userKey(), credentials.caCertificates())) {
+      if (credentials.skipClientAuth()) {
+        return null;
+      }
+      if (anyBlank(credentials.userCertificate(), credentials.userKey())) {
         return "Security protocol " + securityProtocol.name + ": invalid user certificate or user key or CA certificates";
       }
       return null;
@@ -53,9 +56,6 @@ class CredentialsValidator {
     }
 
     if (is(SecurityProtocol.SASL_SSL, securityProtocol)) {
-      if (anyBlank(credentials.caCertificates())) {
-        return "Security protocol " + securityProtocol.name + ": invalid truststore";
-      }
       if (isInvalidSASLMechanism(SASLMechanism)) {
         return "Security protocol " + securityProtocol.name + ": invalid SASL mechanism, expected SCRAM-SHA-256 or SCRAM-SHA-512 got " + SASLMechanism;
       }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
index 31a9cbf32..5ce6de2dd 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
@@ -16,9 +16,6 @@
 
 package dev.knative.eventing.kafka.broker.core.security;
 
-import java.util.Map;
-import java.util.Properties;
-import java.util.function.BiConsumer;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.config.SslConfigs;
@@ -26,6 +23,10 @@ import org.apache.kafka.common.security.plain.PlainLoginModule;
 import org.apache.kafka.common.security.scram.ScramLoginModule;
 import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory;
 
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.BiConsumer;
+
 public class KafkaClientsAuth {
 
   public static Properties attachCredentials(final Properties properties, final Credentials credentials) {
@@ -78,13 +79,17 @@ public class KafkaClientsAuth {
   }
 
   private static void ssl(final BiConsumer<String, String> propertiesSetter, final Credentials credentials) {
-    propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
-    propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, credentials.caCertificates());
-    final var keystore = credentials.userCertificate();
-    if (keystore != null) {
-      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, credentials.userCertificate());
-      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, credentials.userKey());
+    final var caCert = credentials.caCertificates();
+    if (caCert != null) {
+      propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
+      propertiesSetter.accept(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, caCert);
+    }
+    final var userCert = credentials.userCertificate();
+    final var userKey = credentials.userKey();
+    if (userCert != null && userKey != null) {
       propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
+      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, userCert);
+      propertiesSetter.accept(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, userKey);
     }
   }
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java
index 3f065fdd5..0bcd9d7fe 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java
@@ -17,11 +17,12 @@
 package dev.knative.eventing.kafka.broker.core.security;
 
 import io.fabric8.kubernetes.api.model.Secret;
-import java.util.Base64;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Base64;
+
 import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
 class KubernetesCredentials implements Credentials {
@@ -32,6 +33,7 @@ class KubernetesCredentials implements Credentials {
 
   static final String USER_CERTIFICATE_KEY = "user.crt";
   static final String USER_KEY_KEY = "user.key";
+  static final String USER_SKIP_KEY = "user.skip";
 
   static final String USERNAME_KEY = "user";
   static final String PASSWORD_KEY = "password";
@@ -44,6 +46,7 @@ class KubernetesCredentials implements Credentials {
   private String caCertificates;
   private String userCertificate;
   private String userKey;
+  private Boolean skipUser;
   private SecurityProtocol securityProtocol;
   private String SASLMechanism;
   private String SASLUsername;
@@ -68,6 +71,26 @@ class KubernetesCredentials implements Credentials {
     return this.caCertificates;
   }
 
+  @Override
+  public boolean skipClientAuth() {
+    if (secret == null || secret.getData() == null) {
+      return false;
+    }
+    if (skipUser == null) {
+      final var skip = secret.getData().get(USER_SKIP_KEY);
+      if (skip == null) {
+        this.skipUser = false;
+      } else {
+        try {
+          this.skipUser = Boolean.parseBoolean(skip);
+        } catch (final Exception ex) {
+          this.skipUser = false;
+        }
+      }
+    }
+    return this.skipUser;
+  }
+
   @Override
   public String userCertificate() {
     if (secret == null || secret.getData() == null) {
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java
index 2961b5a02..056603c49 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/PlaintextCredentials.java
@@ -25,6 +25,11 @@ public class PlaintextCredentials implements Credentials {
     return null;
   }
 
+  @Override
+  public boolean skipClientAuth() {
+    return true;
+  }
+
   @Override
   public String userCertificate() {
     return null;
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index de45b9426..c343f5da5 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -47,14 +47,14 @@
     <micrometer.version>1.6.4</micrometer.version>
     <opentelemetry.version>0.15.0</opentelemetry.version>
     <jackson.version>2.12.1</jackson.version>
-    <protobuf.version>3.15.0</protobuf.version>
+    <protobuf.version>3.15.3</protobuf.version>
     <slf4j.version>1.7.30</slf4j.version>
     <ch.qos.logback.version>1.2.3</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>6.6</net.logstash.logback.encoder.version>
     <assertj.version>3.19.0</assertj.version>
     <awaitility.version>4.0.3</awaitility.version>
     <junit.jupiter.version>5.7.0</junit.jupiter.version>
-    <mokito.junit.jupiter.version>3.7.7</mokito.junit.jupiter.version>
+    <mokito.junit.jupiter.version>3.8.0</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.0.2</fabric8.kubernetes.version>
     <kafka.version>2.7.0</kafka.version>
     <debezium.version>1.4.1.Final</debezium.version>
diff --git a/go.mod b/go.mod
index 28e53f1f5..2dd235f3c 100644
--- a/go.mod
+++ b/go.mod
@@ -23,7 +23,7 @@ require (
 	k8s.io/apiserver v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a
+	knative.dev/eventing v0.21.1-0.20210226214447-184fa2f5b7ae
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
-	knative.dev/pkg v0.0.0-20210216013737-584933f8280b
+	knative.dev/pkg v0.0.0-20210226182947-9039dc189ced
 )
diff --git a/go.sum b/go.sum
index c37b663df..e69b75096 100644
--- a/go.sum
+++ b/go.sum
@@ -1275,14 +1275,15 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a h1:3cqlUxlZ6M+EZA80OTroRXz7XNNImc4Qix01jqOA1Bo=
-knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a/go.mod h1:JjbVEOTJJHqo9CTxbTfrMn018hG8fOr3UfBoCJ7KWaA=
+knative.dev/eventing v0.21.1-0.20210226214447-184fa2f5b7ae h1:B4qE5DQAnyNiw7EIFS9gRZSg9g9jXIl5Y/ik4roGbEc=
+knative.dev/eventing v0.21.1-0.20210226214447-184fa2f5b7ae/go.mod h1:KtN2FEhPzxyqXKxpJufdq0lzL3Iu+5qjFqBal8Lg2tQ=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf h1:u4cY4jr2LYvhoz/1HBWEPsMiLkm0HMdDTfmmw1RE8zE=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/pkg v0.0.0-20210215165523-84c98f3c3e7a/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/pkg v0.0.0-20210216013737-584933f8280b h1:2v+GBBenjPXhtk3KX/YqxCQVBU6o/AaRkkqE58lgFRQ=
-knative.dev/pkg v0.0.0-20210216013737-584933f8280b/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/reconciler-test v0.0.0-20210216030508-77f50054d024/go.mod h1:RP/K5xJylB72Go6eAsXYEsQHp4zCCNMNjmsqhvq7wko=
+knative.dev/pkg v0.0.0-20210225221047-07b5ddfaea60/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
+knative.dev/pkg v0.0.0-20210226182947-9039dc189ced h1:OCtTbXsJhiR35qCDxFj9Xc8sU3i87ePbc7YlHCt0al4=
+knative.dev/pkg v0.0.0-20210226182947-9039dc189ced/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
+knative.dev/reconciler-test v0.0.0-20210223181346-209f32c8cdc4/go.mod h1:RP/K5xJylB72Go6eAsXYEsQHp4zCCNMNjmsqhvq7wko=
 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/pkg/client/injection/reconciler/eventing/v1/broker/controller.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
index c09035a6d..e4b08753f 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
@@ -117,6 +117,9 @@ func NewImpl(ctx context.Context, r Interface, classValue string, optionsFns ...
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
index d0043b784..606dc0af7 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
@@ -173,6 +173,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/controller.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/controller.go
index f874b63d0..7d5287b7c 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/controller.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/reconciler.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/reconciler.go
index cb6c73de1..7a12aec2a 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/reconciler.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
index 65771bffa..84b5244e4 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
@@ -117,6 +117,9 @@ func NewImpl(ctx context.Context, r Interface, classValue string, optionsFns ...
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
index 1f73a4a43..bfbf932ad 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
@@ -173,6 +173,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
index 06d436648..78ff96730 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
@@ -271,6 +271,9 @@ func NewImpl(ctx {{.contextContext|raw}}, r Interface{{if .hasClass}}, classValu
 			rec.skipStatusUpdates = true
 		}
 		{{- end}}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
index 10149fadd..cc6757da8 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
@@ -363,6 +363,9 @@ func NewReconciler(ctx {{.contextContext|raw}}, logger *{{.zapSugaredLogger|raw}
 			rec.skipStatusUpdates = true
 		}
 		{{- end}}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/vendor/knative.dev/pkg/controller/controller.go b/vendor/knative.dev/pkg/controller/controller.go
index 27fe75e40..2a28497fa 100644
--- a/vendor/knative.dev/pkg/controller/controller.go
+++ b/vendor/knative.dev/pkg/controller/controller.go
@@ -152,6 +152,13 @@ func FilterControllerGK(gk schema.GroupKind) func(obj interface{}) bool {
 	}
 }
 
+// FilterController makes it simple to create FilterFunc's for use with
+// cache.FilteringResourceEventHandler that filter based on the
+// controlling resource.
+func FilterController(r kmeta.OwnerRefable) func(obj interface{}) bool {
+	return FilterControllerGK(r.GetGroupVersionKind().GroupKind())
+}
+
 // FilterWithName makes it simple to create FilterFunc's for use with
 // cache.FilteringResourceEventHandler that filter based on a name.
 func FilterWithName(name string) func(obj interface{}) bool {
diff --git a/vendor/knative.dev/pkg/controller/options.go b/vendor/knative.dev/pkg/controller/options.go
index b40b5a359..8d61835e6 100644
--- a/vendor/knative.dev/pkg/controller/options.go
+++ b/vendor/knative.dev/pkg/controller/options.go
@@ -35,6 +35,9 @@ type Options struct {
 	// SkipStatusUpdates configures this reconciler to either do automated status
 	// updates (default) or skip them if this is set to true.
 	SkipStatusUpdates bool
+
+	// DemoteFunc configures the demote function this reconciler uses
+	DemoteFunc func(b reconciler.Bucket)
 }
 
 // OptionsFn is a callback method signature that accepts an Impl and returns
diff --git a/vendor/knative.dev/pkg/metrics/exporter.go b/vendor/knative.dev/pkg/metrics/exporter.go
index b98a18de1..3fc4966fd 100644
--- a/vendor/knative.dev/pkg/metrics/exporter.go
+++ b/vendor/knative.dev/pkg/metrics/exporter.go
@@ -21,7 +21,6 @@ import (
 	"errors"
 	"fmt"
 	"strings"
-	"sync"
 
 	"go.opencensus.io/resource"
 	"go.opencensus.io/stats/view"
@@ -32,9 +31,14 @@ import (
 var (
 	curMetricsExporter view.Exporter
 	curMetricsConfig   *metricsConfig
-	metricsMux         sync.RWMutex
+	mWorker            *metricsWorker
 )
 
+func init() {
+	mWorker = newMetricsWorker()
+	go mWorker.start()
+}
+
 // SecretFetcher is a function (extracted from SecretNamespaceLister) for fetching
 // a specific Secret. This avoids requiring global or namespace list in controllers.
 type SecretFetcher func(string) (*corev1.Secret, error)
@@ -154,28 +158,14 @@ func UpdateExporter(ctx context.Context, ops ExporterOptions, logger *zap.Sugare
 
 	// Updating the metrics config and the metrics exporters needs to be atomic to
 	// avoid using an outdated metrics config with new exporters.
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-
-	if isNewExporterRequired(newConfig) {
-		logger.Info("Flushing the existing exporter before setting up the new exporter.")
-		flushGivenExporter(curMetricsExporter)
-		e, f, err := newMetricsExporter(newConfig, logger)
-		if err != nil {
-			logger.Errorw("Failed to update a new metrics exporter based on metric config", zap.Error(err), "config", newConfig)
-			return err
-		}
-		existingConfig := curMetricsConfig
-		curMetricsExporter = e
-		if err := setFactory(f); err != nil {
-			logger.Errorw("Failed to update metrics factory when loading metric config", zap.Error(err), "config", newConfig)
-			return err
-		}
-		logger.Infof("Successfully updated the metrics exporter; old config: %v; new config %v", existingConfig, newConfig)
+	updateCmd := &updateMetricsConfigWithExporter{
+		ctx:       ctx,
+		newConfig: newConfig,
+		done:      make(chan error),
 	}
-
-	setCurMetricsConfigUnlocked(newConfig)
-	return nil
+	mWorker.c <- updateCmd
+	err = <-updateCmd.done
+	return err
 }
 
 // isNewExporterRequired compares the non-nil newConfig against curMetricsConfig. When backend changes,
@@ -228,27 +218,35 @@ func newMetricsExporter(config *metricsConfig, logger *zap.SugaredLogger) (view.
 }
 
 func getCurMetricsExporter() view.Exporter {
-	metricsMux.RLock()
-	defer metricsMux.RUnlock()
-	return curMetricsExporter
+	readCmd := &readExporter{done: make(chan *view.Exporter)}
+	mWorker.c <- readCmd
+	e := <-readCmd.done
+	return *e
 }
 
 func setCurMetricsExporter(e view.Exporter) {
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-	curMetricsExporter = e
+	setCmd := &setExporter{
+		newExporter: &e,
+		done:        make(chan struct{}),
+	}
+	mWorker.c <- setCmd
+	<-setCmd.done
 }
 
 func getCurMetricsConfig() *metricsConfig {
-	metricsMux.RLock()
-	defer metricsMux.RUnlock()
-	return curMetricsConfig
+	readCmd := &readMetricsConfig{done: make(chan *metricsConfig)}
+	mWorker.c <- readCmd
+	cfg := <-readCmd.done
+	return cfg
 }
 
 func setCurMetricsConfig(c *metricsConfig) {
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-	setCurMetricsConfigUnlocked(c)
+	setCmd := &setMetricsConfig{
+		newConfig: c,
+		done:      make(chan struct{}),
+	}
+	mWorker.c <- setCmd
+	<-setCmd.done
 }
 
 func setCurMetricsConfigUnlocked(c *metricsConfig) {
diff --git a/vendor/knative.dev/pkg/metrics/metrics_worker.go b/vendor/knative.dev/pkg/metrics/metrics_worker.go
new file mode 100644
index 000000000..0b7753acb
--- /dev/null
+++ b/vendor/knative.dev/pkg/metrics/metrics_worker.go
@@ -0,0 +1,111 @@
+/*
+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 metrics
+
+import (
+	"context"
+
+	"go.opencensus.io/stats/view"
+	"go.uber.org/zap"
+	"knative.dev/pkg/logging"
+)
+
+type metricsWorker struct {
+	c chan command
+}
+
+func newMetricsWorker() *metricsWorker {
+	return &metricsWorker{c: make(chan command)}
+}
+
+type command interface {
+	handleCommand(w *metricsWorker)
+}
+
+type readExporter struct {
+	done chan *view.Exporter
+}
+
+type setExporter struct {
+	newExporter *view.Exporter
+	done        chan struct{}
+}
+
+type readMetricsConfig struct {
+	done chan *metricsConfig
+}
+
+type updateMetricsConfigWithExporter struct {
+	ctx       context.Context
+	newConfig *metricsConfig
+	done      chan error
+}
+
+type setMetricsConfig struct {
+	newConfig *metricsConfig
+	done      chan struct{}
+}
+
+func (cmd *readMetricsConfig) handleCommand(w *metricsWorker) {
+	cmd.done <- curMetricsConfig
+}
+
+func (cmd *setMetricsConfig) handleCommand(w *metricsWorker) {
+	setCurMetricsConfigUnlocked(cmd.newConfig)
+	cmd.done <- struct{}{}
+}
+
+func (cmd *updateMetricsConfigWithExporter) handleCommand(w *metricsWorker) {
+	ctx := cmd.ctx
+	logger := logging.FromContext(ctx)
+	if isNewExporterRequired(cmd.newConfig) {
+		logger.Info("Flushing the existing exporter before setting up the new exporter.")
+		flushGivenExporter(curMetricsExporter)
+		e, f, err := newMetricsExporter(cmd.newConfig, logger)
+		if err != nil {
+			logger.Errorw("Failed to update a new metrics exporter based on metric config", zap.Error(err), "config", cmd.newConfig)
+			cmd.done <- err
+			return
+		}
+		existingConfig := curMetricsConfig
+		curMetricsExporter = e
+		if err := setFactory(f); err != nil {
+			logger.Errorw("Failed to update metrics factory when loading metric config", zap.Error(err), "config", cmd.newConfig)
+			cmd.done <- err
+			return
+		}
+		logger.Infof("Successfully updated the metrics exporter; old config: %v; new config %v", existingConfig, cmd.newConfig)
+	}
+	setCurMetricsConfigUnlocked(cmd.newConfig)
+	cmd.done <- nil
+}
+
+func (w *metricsWorker) start() {
+	for {
+		cmd := <-w.c
+		cmd.handleCommand(w)
+	}
+}
+
+func (cmd *setExporter) handleCommand(w *metricsWorker) {
+	curMetricsExporter = *cmd.newExporter
+	cmd.done <- struct{}{}
+}
+
+func (cmd *readExporter) handleCommand(w *metricsWorker) {
+	cmd.done <- &curMetricsExporter
+}
diff --git a/vendor/knative.dev/pkg/metrics/resource_view.go b/vendor/knative.dev/pkg/metrics/resource_view.go
index 7135af70a..236fd588b 100644
--- a/vendor/knative.dev/pkg/metrics/resource_view.go
+++ b/vendor/knative.dev/pkg/metrics/resource_view.go
@@ -82,10 +82,16 @@ func cleanup() {
 	expiryCutoff := allMeters.clock.Now().Add(-1 * maxMeterExporterAge)
 	allMeters.lock.Lock()
 	defer allMeters.lock.Unlock()
+	resourceViews.lock.Lock()
+	defer resourceViews.lock.Unlock()
 	for key, meter := range allMeters.meters {
 		if key != "" && meter.t.Before(expiryCutoff) {
 			flushGivenExporter(meter.e)
+			// Make a copy of views to avoid data races
+			viewsCopy := copyViews(resourceViews.views)
+			meter.m.Unregister(viewsCopy...)
 			delete(allMeters.meters, key)
+			meter.m.Stop()
 		}
 	}
 }
@@ -139,7 +145,7 @@ func RegisterResourceView(views ...*view.View) error {
 	return nil
 }
 
-// UnregisterResourceView is similar to view.Unregiste(), except that it will
+// UnregisterResourceView is similar to view.Unregister(), except that it will
 // unregister the view across all Resources tracked byt he system, rather than
 // simply the default view.
 func UnregisterResourceView(views ...*view.View) {
diff --git a/vendor/knative.dev/pkg/network/handlers/drain.go b/vendor/knative.dev/pkg/network/handlers/drain.go
index 0ed37ba6d..b8dc6ff1b 100644
--- a/vendor/knative.dev/pkg/network/handlers/drain.go
+++ b/vendor/knative.dev/pkg/network/handlers/drain.go
@@ -17,6 +17,7 @@ limitations under the License.
 package handlers
 
 import (
+	"fmt"
 	"net/http"
 	"sync"
 	"time"
@@ -90,11 +91,11 @@ func (d *Drainer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
-	if network.IsKProbe(r) {
+	if isKProbe(r) {
 		if d.draining() {
 			http.Error(w, "shutting down", http.StatusServiceUnavailable)
 		} else {
-			network.ServeKProbe(w, r)
+			serveKProbe(w, r)
 		}
 		return
 	}
@@ -146,3 +147,21 @@ func (d *Drainer) draining() bool {
 	defer d.RUnlock()
 	return d.timer != nil
 }
+
+// isKProbe returns true if the request is a knatvie probe.
+func isKProbe(r *http.Request) bool {
+	return r.Header.Get(network.ProbeHeaderName) == network.ProbeHeaderValue
+}
+
+// serveKProbe serve KProbe requests.
+func serveKProbe(w http.ResponseWriter, r *http.Request) {
+	hh := r.Header.Get(network.HashHeaderName)
+	if hh == "" {
+		http.Error(w,
+			fmt.Sprintf("a probe request must contain a non-empty %q header", network.HashHeaderName),
+			http.StatusBadRequest)
+		return
+	}
+	w.Header().Set(network.HashHeaderName, hh)
+	w.WriteHeader(http.StatusOK)
+}
diff --git a/vendor/knative.dev/pkg/network/network.go b/vendor/knative.dev/pkg/network/network.go
index b2ac63ce7..cf5feaa8f 100644
--- a/vendor/knative.dev/pkg/network/network.go
+++ b/vendor/knative.dev/pkg/network/network.go
@@ -17,7 +17,6 @@ limitations under the License.
 package network
 
 import (
-	"fmt"
 	"net/http"
 	"strings"
 	"time"
@@ -52,33 +51,19 @@ const (
 	// uses to find out which version of the networking config is deployed.
 	HashHeaderName = "K-Network-Hash"
 
+	// KubeProbeUAPrefix is the prefix for the User-Agent header.
 	// Since K8s 1.8, prober requests have
 	//   User-Agent = "kube-probe/{major-version}.{minor-version}".
 	KubeProbeUAPrefix = "kube-probe/"
 
+	// KubeletProbeHeaderName is the header name to augment the probes, because
 	// Istio with mTLS rewrites probes, but their probes pass a different
-	// user-agent.  So we augment the probes with this header.
+	// user-agent.
 	KubeletProbeHeaderName = "K-Kubelet-Probe"
 )
 
 // IsKubeletProbe returns true if the request is a Kubernetes probe.
 func IsKubeletProbe(r *http.Request) bool {
-	return strings.HasPrefix(r.Header.Get("User-Agent"), KubeProbeUAPrefix) ||
+	return strings.HasPrefix(r.Header.Get(UserAgentKey), KubeProbeUAPrefix) ||
 		r.Header.Get(KubeletProbeHeaderName) != ""
 }
-
-// IsKProbe returns true if the request is a knatvie probe.
-func IsKProbe(r *http.Request) bool {
-	return r.Header.Get(ProbeHeaderName) == ProbeHeaderValue
-}
-
-// ServeKProbe serve KProbe requests.
-func ServeKProbe(w http.ResponseWriter, r *http.Request) {
-	hh := r.Header.Get(HashHeaderName)
-	if hh == "" {
-		http.Error(w, fmt.Sprintf("a probe request must contain a non-empty %q header", HashHeaderName), http.StatusBadRequest)
-		return
-	}
-	w.Header().Set(HashHeaderName, hh)
-	w.WriteHeader(http.StatusOK)
-}
diff --git a/vendor/knative.dev/pkg/system/clock.go b/vendor/knative.dev/pkg/system/clock.go
deleted file mode 100644
index 7d99d9b5c..000000000
--- a/vendor/knative.dev/pkg/system/clock.go
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
-Copyright 2019 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package system
-
-import (
-	"time"
-)
-
-// Mockable interface for time based testing
-type Clock interface {
-	Now() time.Time
-}
-
-type RealClock struct{}
-
-func (RealClock) Now() time.Time {
-	return time.Now()
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index b6ab883d1..16c6557f4 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1014,7 +1014,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a
+# knative.dev/eventing v0.21.1-0.20210226214447-184fa2f5b7ae
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/configs
@@ -1148,7 +1148,7 @@ knative.dev/eventing/test/test_images/request-sender
 # knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210216013737-584933f8280b
+# knative.dev/pkg v0.0.0-20210226182947-9039dc189ced
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/security/secret_test.go b/control-plane/pkg/security/secret_test.go
index f81867aee..537d74b8a 100644
--- a/control-plane/pkg/security/secret_test.go
+++ b/control-plane/pkg/security/secret_test.go
@@ -175,6 +175,85 @@ func TestSSL(t *testing.T) {
 	assert.Greater(t, len(config.Net.TLS.Config.RootCAs.Subjects()), 0)
 }
 
+func TestSSLNoUserKey(t *testing.T) {
+	ca, _, userCert := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol": []byte("SSL"),
+		"user.crt": userCert,
+		"ca.crt":   ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
+func TestSSLNoUserCert(t *testing.T) {
+	ca, userKey, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol": []byte("SSL"),
+		"user.key": userKey,
+		"ca.crt":   ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
+func TestSSLInvalidKeyPair(t *testing.T) {
+	ca, _, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol": []byte("SSL"),
+		"user.key": []byte("foo"),
+		"user.crt": []byte("bar"),
+		"ca.crt":   ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
+func TestSSLNoClientAuth(t *testing.T) {
+	ca, _, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol":  []byte("SSL"),
+		"user.skip": []byte("true"),
+		"ca.crt":    ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.Nil(t, err)
+	assert.True(t, config.Net.TLS.Enable)
+	assert.NotNil(t, config.Net.TLS.Config.RootCAs)
+	assert.Greater(t, len(config.Net.TLS.Config.RootCAs.Subjects()), 0)
+}
+
+func TestSSLNoClientAuthInvalidFlag(t *testing.T) {
+	ca, _, _ := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol":  []byte("SSL"),
+		"user.skip": []byte("foo"),
+		"ca.crt":    ca,
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
 func TestSASLPLainSSL(t *testing.T) {
 	ca, userKey, userCert := loadCerts(t)
 
@@ -259,6 +338,25 @@ func TestSASLSCRAM512SSL(t *testing.T) {
 	assert.Equal(t, "my-user-password", config.Net.SASL.Password)
 }
 
+func TestSASLSCRAM512SSLInvalidCaCert(t *testing.T) {
+	_, userKey, userCert := loadCerts(t)
+
+	secret := map[string][]byte{
+		"protocol":       []byte("SASL_SSL"),
+		"sasl.mechanism": []byte("SCRAM-SHA-512"),
+		"ca.crt":         []byte(`foo`),
+		"user.crt":       userCert,
+		"user.key":       userKey,
+		"user":           []byte("my-user-name"),
+		"password":       []byte("my-user-password"),
+	}
+	config := sarama.NewConfig()
+
+	err := options(config, secretData(secret))
+
+	assert.NotNil(t, err)
+}
+
 func loadCerts(t *testing.T) (ca, userKey, userCert []byte) {
 	ca, err := ioutil.ReadFile("testdata/ca.crt")
 	assert.Nil(t, err)
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
index 248702ba7..1d7b441f4 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
@@ -71,17 +71,31 @@ public class CredentialsValidatorTest {
   }
 
   @Test
-  public void securityProtocolSslInvalidNoCACert() {
+  public void securityProtocolSslNoCACert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SSL);
     when(credential.userCertificate()).thenReturn("xyz");
     when(credential.userKey()).thenReturn("my-key");
-    when(credential.caCertificates()).thenReturn("   ");
+    when(credential.caCertificates()).thenReturn(null);
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
+  @Test
+  public void securityProtocolSslNoCACertNoClientAuth() {
+    final var credential = mock(Credentials.class);
+
+    when(credential.securityProtocol()).thenReturn(SecurityProtocol.SSL);
+    when(credential.skipClientAuth()).thenReturn(true);
+    when(credential.userCertificate()).thenReturn("  ");
+    when(credential.userKey()).thenReturn("  ");
+    when(credential.caCertificates()).thenReturn(null);
+
+    assertThat(CredentialsValidator.validate(credential)).isNull();
+  }
+
+
   @Test
   public void securityProtocolSaslPlaintextScramSha256Valid() {
     securityProtocolSaslPlaintextScramValid("SCRAM-SHA-256");
@@ -195,17 +209,16 @@ public class CredentialsValidatorTest {
   }
 
   @Test
-  public void securityProtocolSaslSslScramSha51NoTruststoreInValid() {
+  public void securityProtocolSaslSslScramSha51EmptyCaCert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
-    when(credential.userCertificate()).thenReturn("abc");
     when(credential.caCertificates()).thenReturn("   ");
     when(credential.SASLMechanism()).thenReturn("SCRAM-SHA-512");
     when(credential.SASLUsername()).thenReturn("bbb");
     when(credential.SASLPassword()).thenReturn("ccc");
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
   @Test
@@ -250,28 +263,29 @@ public class CredentialsValidatorTest {
   }
 
   @Test
-  public void securityProtocolSaslSslScramSha256NoTruststoreInValid() {
+  public void securityProtocolSaslSslScramSha256NoCaCert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
-    when(credential.caCertificates()).thenReturn("   ");
+    when(credential.caCertificates()).thenReturn(null);
     when(credential.SASLMechanism()).thenReturn("SCRAM-SHA-512");
     when(credential.SASLUsername()).thenReturn("bbb");
     when(credential.SASLPassword()).thenReturn("ccc");
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
   @Test
-  public void securityProtocolSaslSslScramSha256NoCACertInValid() {
+  public void securityProtocolSaslSslScramSha256EmptyCACert() {
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
+    when(credential.caCertificates()).thenReturn("  ");
     when(credential.SASLMechanism()).thenReturn("SCRAM-SHA-512");
     when(credential.SASLUsername()).thenReturn("bbb");
     when(credential.SASLPassword()).thenReturn("ccc");
 
-    assertThat(CredentialsValidator.validate(credential)).isNotEmpty();
+    assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
   @Test
diff --git a/vendor/knative.dev/pkg/reconciler/testing/clock.go b/vendor/knative.dev/pkg/reconciler/testing/clock.go
deleted file mode 100644
index 44ba77cdb..000000000
--- a/vendor/knative.dev/pkg/reconciler/testing/clock.go
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
-Copyright 2019 The Knative Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package testing
-
-import (
-	"time"
-)
-
-type FakeClock struct {
-	Time time.Time
-}
-
-func (c FakeClock) Now() time.Time {
-	return c.Time
-}