Repository: knative-sandbox/eventing-kafka-broker
Pull request #669 (2021-02-21 20:54:40)

Move reconcileKind method of Trigger reconciler
The reconcileKind method of the Trigger reconciler was below
other minor methods, this PR moves it above, no logic changed.

## Proposed Changes

- Move reconcileKind method of Trigger reconciler

/kind cleanup

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/669?src=pr&el=h1) Report
> Merging [#669](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669?src=pr&el=desc) (6b15bb2) 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 `67.85%`.

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

```diff
@@             Coverage Diff              @@
##             master     #669      +/-   ##
============================================
- 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/669?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [control-plane/pkg/reconciler/trigger/trigger.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXIuZ28=) | `71.20% <67.85%> (ø)` | `0.00 <0.00> (ø)` | |
| [...spatcher/consumer/impl/UnorderedOffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvY29uc3VtZXIvaW1wbC9Vbm9yZGVyZWRPZmZzZXRNYW5hZ2VyLmphdmE=) | | | |
| [...ve/eventing/kafka/broker/core/tracing/Tracing.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9UcmFjaW5nLmphdmE=) | | | |
| [...oker/dispatcher/http/IllegalResponseException.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaHR0cC9JbGxlZ2FsUmVzcG9uc2VFeGNlcHRpb24uamF2YQ==) | | | |
| [...a/broker/core/security/KubernetesAuthProvider.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS3ViZXJuZXRlc0F1dGhQcm92aWRlci5qYXZh) | | | |
| [...afka/broker/core/filter/impl/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsdGVyL2ltcGwvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | | |
| [.../broker/dispatcher/consumer/DeliveryGuarantee.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvY29uc3VtZXIvRGVsaXZlcnlHdWFyYW50ZWUuamF2YQ==) | | | |
| [...ve/eventing/kafka/broker/receiver/ReceiverEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9SZWNlaXZlckVudi5qYXZh) | | | |
| [...nting/kafka/broker/core/tracing/TracingConfig.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9UcmFjaW5nQ29uZmlnLmphdmE=) | | | |
| [...r/dispatcher/http/HttpConsumerVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaHR0cC9IdHRwQ29uc3VtZXJWZXJ0aWNsZUZhY3RvcnkuamF2YQ==) | | | |
| ... and [40 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669?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/669?src=pr&el=footer). Last update [182da5c...6b15bb2](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/669?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/reconciler/kafka/topic.go b/control-plane/pkg/reconciler/kafka/topic.go
index bbabef261..8423ac13e 100644
--- a/control-plane/pkg/reconciler/kafka/topic.go
+++ b/control-plane/pkg/reconciler/kafka/topic.go
@@ -43,7 +43,18 @@ func BootstrapServersCommaSeparated(bootstrapServers []string) string {
 }
 
 func BootstrapServersArray(bootstrapServers string) []string {
-	return strings.Split(bootstrapServers, ",")
+	bss := strings.Split(bootstrapServers, ",")
+
+	j := len(bss)
+	for i := 0; i < j; i++ {
+		bss[i] = strings.TrimSpace(bss[i])
+		if bss[i] == "" {
+			j--
+			bss[i] = bss[j]
+			i--
+		}
+	}
+	return bss[:j]
 }
 
 // Topic returns a topic name given a topic prefix and a generic object.
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index 49b257502..b381b03fa 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -56,6 +56,116 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, trigger *eventing.Trigge
 	})
 }
 
+func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
+
+	logger := log.Logger(ctx, "reconcile", trigger)
+
+	statusConditionManager := statusConditionManager{
+		Trigger:  trigger,
+		Configs:  r.Configs,
+		Recorder: controller.GetEventRecorder(ctx),
+	}
+
+	broker, err := r.BrokerLister.Brokers(trigger.Namespace).Get(trigger.Spec.Broker)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return statusConditionManager.failedToGetBroker(err)
+	}
+
+	if apierrors.IsNotFound(err) {
+
+		// Actually check if the broker doesn't exist.
+		// Note: do not introduce another `broker` variable with `:`
+		broker, err = r.EventingClient.EventingV1().Brokers(trigger.Namespace).Get(ctx, trigger.Spec.Broker, metav1.GetOptions{})
+
+		if apierrors.IsNotFound(err) {
+
+			logger.Debug("broker not found", zap.String("finalizeDuringReconcile", "notFound"))
+			// The associated broker doesn't exist anymore, so clean up Trigger resources.
+			return r.FinalizeKind(ctx, trigger)
+		}
+	}
+
+	// Ignore Triggers that are associated with a Broker we don't own.
+	if isOur, brokerClass := isOurBroker(broker); !isOur {
+		logger.Debug("Ignoring Trigger", zap.String(eventing.BrokerClassAnnotationKey, brokerClass))
+		return nil
+	}
+
+	if !broker.GetDeletionTimestamp().IsZero() {
+
+		logger.Debug("broker deleted", zap.String("finalizeDuringReconcile", "deleted"))
+
+		// The associated broker doesn't exist anymore, so clean up Trigger resources.
+		return r.FinalizeKind(ctx, trigger)
+	}
+
+	statusConditionManager.propagateBrokerCondition(broker)
+
+	// Get data plane config map.
+	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return statusConditionManager.failedToGetDataPlaneConfigMap(err)
+	}
+
+	logger.Debug("Got contract config map")
+
+	// Get data plane config data.
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	if err != nil || ct == nil {
+		return statusConditionManager.failedToGetDataPlaneConfigFromConfigMap(err)
+	}
+
+	logger.Debug(
+		"Got contract data from config map",
+		zap.Any(base.ContractLogKey, (*log.ContractMarshaller)(ct)),
+	)
+
+	brokerIndex := coreconfig.FindResource(ct, broker.UID)
+	if brokerIndex == coreconfig.NoResource {
+		return statusConditionManager.brokerNotFoundInDataPlaneConfigMap()
+	}
+	triggerIndex := coreconfig.FindEgress(ct.Resources[brokerIndex].Egresses, trigger.UID)
+
+	triggerConfig, err := r.getTriggerConfig(ctx, trigger)
+	if err != nil {
+		return statusConditionManager.failedToResolveTriggerConfig(err)
+	}
+	statusConditionManager.subscriberResolved()
+
+	changed := coreconfig.AddOrUpdateEgressConfig(ct, brokerIndex, triggerConfig, triggerIndex)
+
+	coreconfig.IncrementContractGeneration(ct)
+
+	logger.Debug("Egress changes", zap.Int("changed", changed))
+
+	if changed == coreconfig.EgressChanged {
+		// Update the configuration map with the new dataPlaneConfig data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			trigger.Status.MarkDependencyFailed(string(base.ConditionConfigMapUpdated), err.Error())
+			return err
+		}
+
+		// Update volume generation annotation of dispatcher pods
+		if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+			// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+			// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Trigger
+			// ready. So, log out the error and move on to the next step.
+			logger.Warn(
+				"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+				zap.Error(err),
+			)
+
+			statusConditionManager.failedToUpdateDispatcherPodsAnnotation(err)
+		} else {
+			logger.Debug("Updated dispatcher pod annotation")
+		}
+	}
+
+	logger.Debug("Contract config map updated")
+
+	return statusConditionManager.reconciled()
+}
+
 func (r *Reconciler) FinalizeKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
 	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
 		return r.finalizeKind(ctx, trigger)
@@ -183,116 +293,6 @@ func deleteTrigger(egresses []*contract.Egress, index int) []*contract.Egress {
 	return egresses[:len(egresses)-1]
 }
 
-func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
-
-	logger := log.Logger(ctx, "reconcile", trigger)
-
-	statusConditionManager := statusConditionManager{
-		Trigger:  trigger,
-		Configs:  r.Configs,
-		Recorder: controller.GetEventRecorder(ctx),
-	}
-
-	broker, err := r.BrokerLister.Brokers(trigger.Namespace).Get(trigger.Spec.Broker)
-	if err != nil && !apierrors.IsNotFound(err) {
-		return statusConditionManager.failedToGetBroker(err)
-	}
-
-	if apierrors.IsNotFound(err) {
-
-		// Actually check if the broker doesn't exist.
-		// Note: do not introduce another `broker` variable with `:`
-		broker, err = r.EventingClient.EventingV1().Brokers(trigger.Namespace).Get(ctx, trigger.Spec.Broker, metav1.GetOptions{})
-
-		if apierrors.IsNotFound(err) {
-
-			logger.Debug("broker not found", zap.String("finalizeDuringReconcile", "notFound"))
-			// The associated broker doesn't exist anymore, so clean up Trigger resources.
-			return r.FinalizeKind(ctx, trigger)
-		}
-	}
-
-	// Ignore Triggers that are associated with a Broker we don't own.
-	if isOur, brokerClass := isOurBroker(broker); !isOur {
-		logger.Debug("Ignoring Trigger", zap.String(eventing.BrokerClassAnnotationKey, brokerClass))
-		return nil
-	}
-
-	if !broker.GetDeletionTimestamp().IsZero() {
-
-		logger.Debug("broker deleted", zap.String("finalizeDuringReconcile", "deleted"))
-
-		// The associated broker doesn't exist anymore, so clean up Trigger resources.
-		return r.FinalizeKind(ctx, trigger)
-	}
-
-	statusConditionManager.propagateBrokerCondition(broker)
-
-	// Get data plane config map.
-	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
-	if err != nil {
-		return statusConditionManager.failedToGetDataPlaneConfigMap(err)
-	}
-
-	logger.Debug("Got contract config map")
-
-	// Get data plane config data.
-	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
-	if err != nil || ct == nil {
-		return statusConditionManager.failedToGetDataPlaneConfigFromConfigMap(err)
-	}
-
-	logger.Debug(
-		"Got contract data from config map",
-		zap.Any(base.ContractLogKey, (*log.ContractMarshaller)(ct)),
-	)
-
-	brokerIndex := coreconfig.FindResource(ct, broker.UID)
-	if brokerIndex == coreconfig.NoResource {
-		return statusConditionManager.brokerNotFoundInDataPlaneConfigMap()
-	}
-	triggerIndex := coreconfig.FindEgress(ct.Resources[brokerIndex].Egresses, trigger.UID)
-
-	triggerConfig, err := r.getTriggerConfig(ctx, trigger)
-	if err != nil {
-		return statusConditionManager.failedToResolveTriggerConfig(err)
-	}
-	statusConditionManager.subscriberResolved()
-
-	changed := coreconfig.AddOrUpdateEgressConfig(ct, brokerIndex, triggerConfig, triggerIndex)
-
-	coreconfig.IncrementContractGeneration(ct)
-
-	logger.Debug("Egress changes", zap.Int("changed", changed))
-
-	if changed == coreconfig.EgressChanged {
-		// Update the configuration map with the new dataPlaneConfig data.
-		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
-			trigger.Status.MarkDependencyFailed(string(base.ConditionConfigMapUpdated), err.Error())
-			return err
-		}
-
-		// Update volume generation annotation of dispatcher pods
-		if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-			// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
-			// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Trigger
-			// ready. So, log out the error and move on to the next step.
-			logger.Warn(
-				"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
-				zap.Error(err),
-			)
-
-			statusConditionManager.failedToUpdateDispatcherPodsAnnotation(err)
-		} else {
-			logger.Debug("Updated dispatcher pod annotation")
-		}
-	}
-
-	logger.Debug("Contract config map updated")
-
-	return statusConditionManager.reconciled()
-}
-
 func isOurBroker(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
 	return brokerClass == kafka.BrokerClass, brokerClass
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/reconciler/base/reconciler_test.go b/control-plane/pkg/reconciler/base/reconciler_test.go
index ec424cb0f..decde42c3 100644
--- a/control-plane/pkg/reconciler/base/reconciler_test.go
+++ b/control-plane/pkg/reconciler/base/reconciler_test.go
@@ -14,23 +14,227 @@
  * limitations under the License.
  */
 
-package base
+package base_test
 
 import (
+	"context"
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/encoding/protojson"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	"knative.dev/pkg/logging"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
+func TestIsReceiverRunning(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	addRunningPod(podinformer.Get(ctx).Informer().GetStore(), kubeclient.Get(ctx), base.BrokerReceiverLabel)
+
+	r := &base.Reconciler{
+		PodLister:     podinformer.Get(ctx).Lister(),
+		ReceiverLabel: base.BrokerReceiverLabel,
+	}
+
+	require.True(t, r.IsReceiverRunning())
+}
+
+func TestIsReceiverNotRunning(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	r := &base.Reconciler{
+		PodLister:     podinformer.Get(ctx).Lister(),
+		ReceiverLabel: base.BrokerReceiverLabel,
+	}
+
+	require.False(t, r.IsReceiverRunning())
+}
+
+func TestIsDispatcherRunning(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	label := "dispatcher"
+
+	addRunningPod(podinformer.Get(ctx).Informer().GetStore(), kubeclient.Get(ctx), label)
+
+	r := &base.Reconciler{
+		PodLister:       podinformer.Get(ctx).Lister(),
+		DispatcherLabel: label,
+	}
+
+	require.True(t, r.IsDispatcherRunning())
+}
+
+func TestIsDispatcherNotRunning(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	r := &base.Reconciler{
+		PodLister:       podinformer.Get(ctx).Lister(),
+		DispatcherLabel: base.BrokerDispatcherLabel,
+	}
+
+	require.False(t, r.IsDispatcherRunning())
+}
+
+func TestGetOrCreateDataPlaneConfigMap(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	r := &base.Reconciler{
+		KubeClient: kubeclient.Get(ctx),
+	}
+
+	cm, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	require.Nil(t, err)
+	require.NotNil(t, cm)
+}
+
+func TestGetDataPlaneConfigMapDataEmptyConfigMap(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	r := &base.Reconciler{
+		KubeClient: kubeclient.Get(ctx),
+	}
+
+	cm := &corev1.ConfigMap{}
+
+	ct, err := r.GetDataPlaneConfigMapData(logging.FromContext(ctx).Desugar(), cm)
+	require.Nil(t, err)
+	require.NotNil(t, ct)
+	require.Equal(t, uint64(0), ct.Generation)
+	require.Len(t, ct.Resources, 0)
+}
+
+func TestGetDataPlaneConfigMapData(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	r := &base.Reconciler{
+		KubeClient:            kubeclient.Get(ctx),
+		DataPlaneConfigFormat: base.Json,
+	}
+
+	ct := &contract.Contract{}
+	ct.Resources = append(ct.Resources, &contract.Resource{Uid: "123"})
+
+	b, err := protojson.Marshal(ct)
+	require.Nil(t, err)
+
+	cm := &corev1.ConfigMap{
+		BinaryData: map[string][]byte{
+			base.ConfigMapDataKey: b,
+		},
+	}
+
+	got, err := r.GetDataPlaneConfigMapData(logging.FromContext(ctx).Desugar(), cm)
+	require.Nil(t, err)
+	require.NotNil(t, got)
+	require.Len(t, got.Resources, len(ct.Resources))
+
+	ctStr, err := protojson.Marshal(ct)
+	require.Nil(t, err)
+	gotStr, err := protojson.Marshal(got)
+	require.Nil(t, err)
+
+	if diff := cmp.Diff(ctStr, gotStr); diff != "" {
+		t.Fatal("(-want, +got)", diff)
+	}
+}
+
+func TestUpdateDataPlaneConfigMap(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "ns",
+			Name:      "name",
+		},
+		BinaryData: map[string][]byte{base.ConfigMapDataKey: []byte("")},
+	}
+
+	_, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(cm.Namespace).Create(ctx, cm, metav1.CreateOptions{})
+	require.Nil(t, err)
+
+	r := &base.Reconciler{
+		KubeClient:            kubeclient.Get(ctx),
+		DataPlaneConfigFormat: base.Json,
+	}
+
+	ct := &contract.Contract{}
+	ct.Resources = append(ct.Resources, &contract.Resource{Uid: "123"})
+
+	err = r.UpdateDataPlaneConfigMap(ctx, ct, cm)
+	require.Nil(t, err)
+}
+
+func TestGetDataPlaneConfigMapDataCorrupted(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	r := &base.Reconciler{
+		KubeClient:            kubeclient.Get(ctx),
+		DataPlaneConfigFormat: base.Json,
+	}
+
+	cm := &corev1.ConfigMap{
+		BinaryData: map[string][]byte{
+			base.ConfigMapDataKey: []byte("corrupted"),
+		},
+	}
+
+	got, err := r.GetDataPlaneConfigMapData(logging.FromContext(ctx).Desugar(), cm)
+	require.NotNil(t, err)
+	require.Equal(t, uint64(0), got.Generation)
+}
+
+func TestUpdateReceiverPodAnnotation(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	addRunningPod(podinformer.Get(ctx).Informer().GetStore(), kubeclient.Get(ctx), base.SinkReceiverLabel)
+
+	r := &base.Reconciler{
+		PodLister:     podinformer.Get(ctx).Lister(),
+		KubeClient:    kubeclient.Get(ctx),
+		ReceiverLabel: base.SinkReceiverLabel,
+	}
+
+	err := r.UpdateReceiverPodsAnnotation(ctx, logging.FromContext(ctx).Desugar(), 1)
+	require.Nil(t, err)
+}
+
+func TestUpdateDispatcherPodAnnotation(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	label := "dispatcher"
+
+	addRunningPod(podinformer.Get(ctx).Informer().GetStore(), kubeclient.Get(ctx), label)
+
+	r := &base.Reconciler{
+		PodLister:       podinformer.Get(ctx).Lister(),
+		KubeClient:      kubeclient.Get(ctx),
+		DispatcherLabel: label,
+	}
+
+	err := r.UpdateDispatcherPodsAnnotation(ctx, logging.FromContext(ctx).Desugar(), 1)
+	require.Nil(t, err)
+}
+
 func TestTrackConfigMap(t *testing.T) {
 
-	r := &Reconciler{
+	r := &base.Reconciler{
 		ConfigMapTracker: tracker.New(func(name types.NamespacedName) {}, time.Second),
 	}
 
@@ -47,3 +251,46 @@ func TestTrackConfigMap(t *testing.T) {
 	err := r.TrackConfigMap(cm, &eventing.Broker{})
 	assert.Nil(t, err)
 }
+
+func TestTrackSecret(t *testing.T) {
+
+	r := &base.Reconciler{
+		SecretTracker: tracker.New(func(name types.NamespacedName) {}, time.Second),
+	}
+
+	cm := &corev1.Secret{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ConfigMap",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "my-namespace",
+			Name:      "my-name",
+		},
+	}
+	err := r.TrackSecret(cm, &eventing.Broker{})
+	assert.Nil(t, err)
+}
+
+func TestOnDeleteObserver(t *testing.T) {
+
+}
+
+func addRunningPod(store cache.Store, kc kubernetes.Interface, label string) {
+	pod := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod",
+			Namespace: "ns",
+			Labels:    map[string]string{"app": label},
+		},
+		Status: corev1.PodStatus{Phase: corev1.PodRunning},
+	}
+
+	if err := store.Add(pod); err != nil {
+		panic(err)
+	}
+	if _, err := kc.CoreV1().Pods("ns").Create(context.Background(), pod, metav1.CreateOptions{}); err != nil {
+		panic(err)
+	}
+}
diff --git a/control-plane/pkg/reconciler/kafka/admin_test.go b/control-plane/pkg/reconciler/kafka/admin_test.go
new file mode 100644
index 000000000..878863c1a
--- /dev/null
+++ b/control-plane/pkg/reconciler/kafka/admin_test.go
@@ -0,0 +1,61 @@
+/*
+ * 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 kafka
+
+import (
+	"io"
+	"testing"
+
+	"github.com/Shopify/sarama"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
+)
+
+func TestGetClusterAdmin(t *testing.T) {
+
+	tests := []struct {
+		name             string
+		adminFunc        NewClusterAdminFunc
+		bootstrapServers []string
+		secOptions       security.ConfigOption
+		wantErr          bool
+	}{
+		{
+			name:             "security options error",
+			adminFunc:        func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) { return nil, nil },
+			bootstrapServers: []string{"bs:9092"},
+			secOptions:       func(config *sarama.Config) error { return io.EOF },
+			wantErr:          true,
+		},
+		{
+			name:             "admin func error",
+			adminFunc:        func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) { return nil, io.EOF },
+			bootstrapServers: []string{"bs:9092"},
+			secOptions:       func(config *sarama.Config) error { return nil },
+			wantErr:          true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := GetClusterAdmin(tt.adminFunc, tt.bootstrapServers, tt.secOptions)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("GetClusterAdmin() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/kafka/topic_test.go b/control-plane/pkg/reconciler/kafka/topic_test.go
index 69d34875a..b2884fa7c 100644
--- a/control-plane/pkg/reconciler/kafka/topic_test.go
+++ b/control-plane/pkg/reconciler/kafka/topic_test.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -438,3 +439,12 @@ func TestNewClusterAdminFuncIsTopicPresentCloseClusterAdmin(t *testing.T) {
 	assert.True(t, got, "IsTopicPresentAndValid() got = %v, want %v", got, true)
 	assert.True(t, ca.ExpectedClose, "expected call to Close() on ClusterAdmin")
 }
+
+func TestBootstrapServersArray(t *testing.T) {
+	bss := BootstrapServersArray("bs:9091, bs:9000,,bs:9002,")
+
+	require.Contains(t, bss, "bs:9091")
+	require.Contains(t, bss, "bs:9000")
+	require.Contains(t, bss, "bs:9002")
+	require.Len(t, bss, 3)
+}
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
-}