Repository: knative-sandbox/eventing-kafka
Pull request #284 (2020-12-17 02:29:12)

[master] Upgrade to latest dependencies
Produced via:
```shell
./hack/update-deps.sh --upgrade
./hack/update-codegen.sh
```
Cron. -knative-prow-robot
/cc knative-sandbox/channel-wg-leads
/assign knative-sandbox/channel-wg-leads

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/284#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **grantr** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @grantr` 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).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/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":["grantr"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/284?src=pr&el=h1) Report
> Merging [#284](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/284?src=pr&el=desc) (3ae35f9) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/0054be80bf8e288ae9b806d1d66ddf1bf54f0902?el=desc) (0054be8) will **not change** coverage.
> The diff coverage is `n/a`.

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

```diff
@@           Coverage Diff           @@
##           master     #284   +/-   ##
=======================================
  Coverage   75.63%   75.63%           
=======================================
  Files         114      114           
  Lines        4474     4474           
=======================================
  Hits         3384     3384           
  Misses        888      888           
  Partials      202      202           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/284?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/pull/284?src=pr&el=footer). Last update [0054be8...3ae35f9](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/284?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/284#" title="Author self-approved">knative-automation</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/284#pullrequestreview-554339948" title="Approved">matzew</a>*

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

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

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

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

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":[]} -->

Patch:
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index b38122ab6..3c7f28dde 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -209,54 +209,6 @@ jobs:
 
           echo '::endgroup::'
 
-      - name: Redundant Format
-        shell: bash
-        if: ${{ always() }}
-        env:
-          REVIEWDOG_GITHUB_API_TOKEN: ${{ github.token }}
-        run: |
-          set -e
-          cd "${GITHUB_WORKSPACE}" || exit 1
-
-          echo '::group:: Flagging possibly unnecessary format calls for trailing argument with reviewdog 🐶 ...'
-          # Don't fail because of misspell
-          set +o pipefail
-          # For all look for formatting calls and extract the ones that
-          # have only a single formatting argument and that argument is in the last position.
-          # But exclude the `fmt.Errorf` calls since there's no `fmt.Error` and `errors.New` is not
-          # as versatile.
-          # Also ignore `%T` — to print the type and `%q` to quote the final string.
-          fn=$(find . -path './vendor' -prune -o -path './third_party' -prune -o -name '*.pb.go' -prune -o -type f -name '*.go' -print)
-          grep -nE '[^.]+\.(Fatalf|Errorf|Warnf|Infof|Debugf|Logf|Sprintf|Fprintf|Printf)[^\%]+%[^Tqx]",[^,]+' $fn | grep -v "fmt.Errorf" |
-            while read -r ent ; do
-              file=$(echo $ent | cut -d':' -f 1);
-              line=$(echo $ent | cut -d':' -f 2);
-              ch=$(echo $ent | cut -d':' -f3-);
-              err="Unknown printer tool, please file an issue in knative-sandbox/.github and assign to @vagababov: $ch"
-              if echo $ch | grep --quiet -E "^t.(Errorf|Fatalf|Logf)" ; then
-                err=$(echo $ch | sed  -E 's/(^t\.)(Fatal|Error|Log)f([^\%]+)( %[^Tqx]",)([^,]+)/\1\2\3",\5/')
-                # Not a test. Here we deal with various loggers and fmt helpers.
-              elif echo $ch | grep --quiet "log" ; then
-                # Capture (x)?log(er)?.
-                err=$(echo $ch | sed -E 's/(.*log.*\.)(Print|Fatal|Error|Info|Warn)f([^\%]+)(%[^Tq]",)([^,]+)/\1\2\3",\5/')
-              elif echo $ch | grep --quiet -E "fmt\.Sprintf" ; then
-                # Always space after sprintf
-                err=$(echo $ch | sed -E 's/(fmt\.)(Sprint)f([^%]+) (%s",)([^,]+)/\1\2\3 ",\5/')
-              elif echo $ch | grep --quiet -E "fmt\." ; then  # all other fmt. printers.
-                err=$(echo $ch | sed -E 's/(fmt\.)(Print|Fprint)f([^%]+) (%[^sTxq]",)([^,]+)/\1\2\3",\5/')
-              fi
-              # Trim spaces.
-              ch=$(echo $ch | xargs)
-              echo "$file:$line: Please consider avoiding tail format like this:%0A- $ch%0A+ $err"
-          done |
-          reviewdog -efm="%f:%l: %m" \
-                -name="Redundant Format" \
-                -reporter="github-pr-check" \
-                -filter-mode="added" \
-                -fail-on-error="true" \
-                -level="error"
-          echo '::endgroup::'
-
       # This is mostly copied from https://github.com/get-woke/woke-action-reviewdog/blob/main/entrypoint.sh
       # since their action is not yet released under a stable version.
       - name: Language
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index bd71f1764..9b1df2857 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -37,6 +37,7 @@ import (
 	dispatcherhealth "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/health"
 	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	kncontroller "knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	eventingmetrics "knative.dev/pkg/metrics"
@@ -77,7 +78,7 @@ func main() {
 	}
 
 	// Update The Sarama Config - Username/Password Overrides (EnvVars From Secret Take Precedence Over ConfigMap)
-	sarama.UpdateSaramaConfig(saramaConfig, constants.Component, environment.KafkaUsername, environment.KafkaPassword)
+	client.UpdateSaramaConfig(saramaConfig, constants.Component, environment.KafkaUsername, environment.KafkaPassword)
 
 	// Enable Sarama Logging If Specified In ConfigMap
 	sarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 0d317c4b8..b7c03b4ef 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -37,6 +37,7 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/env"
 	channelhealth "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/producer"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	eventingchannel "knative.dev/eventing/pkg/channel"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/logging"
@@ -80,7 +81,7 @@ func main() {
 	}
 
 	// Update The Sarama Config - Username/Password Overrides (EnvVars From Secret Take Precedence Over ConfigMap)
-	sarama.UpdateSaramaConfig(saramaConfig, constants.Component, environment.KafkaUsername, environment.KafkaPassword)
+	client.UpdateSaramaConfig(saramaConfig, constants.Component, environment.KafkaUsername, environment.KafkaPassword)
 
 	// Enable Sarama Logging If Specified In ConfigMap
 	sarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
diff --git a/go.mod b/go.mod
index 97e9d35ab..6b6303661 100644
--- a/go.mod
+++ b/go.mod
@@ -32,9 +32,9 @@ require (
 	k8s.io/apimachinery v0.18.12
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
-	knative.dev/eventing v0.19.1-0.20201216081458-3e06c6b996dd
+	knative.dev/eventing v0.19.1-0.20201216183558-f3468d1c9954
 	knative.dev/hack v0.0.0-20201214230143-4ed1ecb8db24
-	knative.dev/pkg v0.0.0-20201216014558-95b8793bd623
+	knative.dev/pkg v0.0.0-20201217014702-b0c121fb667f
 )
 
 replace (
diff --git a/go.sum b/go.sum
index 843d0970d..390560bda 100644
--- a/go.sum
+++ b/go.sum
@@ -1275,8 +1275,8 @@ k8s.io/kube-openapi v0.0.0-20200410145947-bcb3869e6f29/go.mod h1:F+5wygcW0wmRTnM
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7MpmSnvtrOieolJKoE=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.19.1-0.20201216081458-3e06c6b996dd h1:5qYvQl79MwwIaFK8FT86CfCgQok3BFBKta8CPbpDaZ0=
-knative.dev/eventing v0.19.1-0.20201216081458-3e06c6b996dd/go.mod h1:Zs0qfFfbKDhEM5JHjQtmCXBu9k+GERSfeDwitMtgjQM=
+knative.dev/eventing v0.19.1-0.20201216183558-f3468d1c9954 h1:6aNIdsIusWwC2esDUuvzSqyf7hUPohwsX2avcR/q0Kc=
+knative.dev/eventing v0.19.1-0.20201216183558-f3468d1c9954/go.mod h1:Zs0qfFfbKDhEM5JHjQtmCXBu9k+GERSfeDwitMtgjQM=
 knative.dev/hack v0.0.0-20201112185459-01a34c573bd8 h1:RNbZsAjhswBPtl4C5C5gEFX5/GfWIOZQxfYD9DhkHdY=
 knative.dev/hack v0.0.0-20201112185459-01a34c573bd8/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20201118155651-b31d3bb6bff9 h1:OgYL1cATsWTS6kztSOuQqhmYfcn6ZdpPAQQ6TKCx00I=
@@ -1286,8 +1286,8 @@ knative.dev/hack v0.0.0-20201214230143-4ed1ecb8db24/go.mod h1:PHt8x8yX5Z9pPquBEf
 knative.dev/pkg v0.0.0-20201117221452-0fccc54273ed h1:4LNEFhvD2Ya4WgCD3SC1WGXz81bjEvQgbcpe5XACI3I=
 knative.dev/pkg v0.0.0-20201117221452-0fccc54273ed/go.mod h1:nxlh3CUvx6WBPr1WKD96AHxFZPD2UKRDo9RUp8ILTyQ=
 knative.dev/pkg v0.0.0-20201215202458-ef8048c0ba77/go.mod h1:VjrwVhfEZUnn6FruncHcBm854FldnRpekpyBbYtBvZM=
-knative.dev/pkg v0.0.0-20201216014558-95b8793bd623 h1:YDt8T3at2oWi33WD3UzbtE9bh+r8iQ0xJijJc7k7nqw=
-knative.dev/pkg v0.0.0-20201216014558-95b8793bd623/go.mod h1:VjrwVhfEZUnn6FruncHcBm854FldnRpekpyBbYtBvZM=
+knative.dev/pkg v0.0.0-20201217014702-b0c121fb667f h1:c4fPgRAIlW0uTC2Ec3+AfkA3adPXwfjfA0NhiA/ybgs=
+knative.dev/pkg v0.0.0-20201217014702-b0c121fb667f/go.mod h1:VjrwVhfEZUnn6FruncHcBm854FldnRpekpyBbYtBvZM=
 knative.dev/reconciler-test v0.0.0-20201124190335-83a44efcdfef/go.mod h1:YSs1y1rgnjs8w39/drLIOQbWvZUQwqApvd+EizO8UsA=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/pkg/channel/distributed/common/kafka/admin/admin_kafka.go b/pkg/channel/distributed/common/kafka/admin/admin_kafka.go
index f4d6a52e7..88c60cf94 100644
--- a/pkg/channel/distributed/common/kafka/admin/admin_kafka.go
+++ b/pkg/channel/distributed/common/kafka/admin/admin_kafka.go
@@ -27,7 +27,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	adminutil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/util"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
-	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 )
@@ -87,7 +87,7 @@ func NewKafkaAdminClient(ctx context.Context, saramaConfig *sarama.Config, clien
 	password := string(kafkaSecret.Data[constants.KafkaSecretKeyPassword])
 
 	// Update The Sarama ClusterAdmin Configuration With Our Values
-	kafkasarama.UpdateSaramaConfig(saramaConfig, clientId, username, password)
+	client.UpdateSaramaConfig(saramaConfig, clientId, username, password)
 
 	// Create A New Sarama ClusterAdmin
 	clusterAdmin, err := NewClusterAdminWrapper(brokers, saramaConfig)
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama.go b/pkg/channel/distributed/common/kafka/sarama/sarama.go
index 6ebc55561..4bd527b0e 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama.go
@@ -24,7 +24,6 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
-	"regexp"
 
 	"github.com/Shopify/sarama"
 	"github.com/ghodss/yaml"
@@ -33,15 +32,12 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/system"
 )
 
-// Regular Expression To Find All Certificates In Net.TLS.Config.RootPEMs Field
-var regexRootPEMs = regexp.MustCompile(`(?s)\s*RootPEMs:.*-----END CERTIFICATE-----`)
-
 // Utility Function For Enabling Sarama Logging (Debugging)
 func EnableSaramaLogging(enable bool) {
 	if enable {
@@ -51,156 +47,6 @@ func EnableSaramaLogging(enable bool) {
 	}
 }
 
-// Utility Function For Configuring Common Settings For Admin/Producer/Consumer
-func UpdateSaramaConfig(config *sarama.Config, clientId string, username string, password string) {
-
-	// Set The ClientID For Logging
-	config.ClientID = clientId
-
-	// Set The SASL Username / Password
-	config.Net.SASL.User = username
-	config.Net.SASL.Password = password
-
-	// We Always Want To Know About Consumer Errors
-	config.Consumer.Return.Errors = true
-
-	// We Always Want Success Messages From Producer
-	config.Producer.Return.Successes = true
-}
-
-//
-// Extract (Parse & Remove) Top Level Kafka Version From Specified Sarama Confirm YAML String
-//
-// The Sarama.Config struct contains a top-level 'Version' field of type sarama.KafkaVersion.
-// This type contains a package scoped [4]uint field called 'version' which we cannot unmarshal
-// the yaml into.  Therefore, we instead support the user providing a 'Version' string of the
-// format "major.minor.patch" (e.g. "2.3.0") which we will "extract" out of the YAML string
-// and return as an actual sarama.KafkaVersion.  In the case where the user has NOT specified
-// a Version string we will return a default version.
-//
-func extractKafkaVersion(saramaConfigYamlString string) (string, sarama.KafkaVersion, error) {
-
-	// Define Inline Struct To Marshall The Top Level Sarama Config "Kafka" Version Into
-	type saramaConfigShell struct {
-		Version string
-	}
-
-	// Unmarshal The Sarama Config Into The Shell
-	shell := &saramaConfigShell{}
-	err := yaml.Unmarshal([]byte(saramaConfigYamlString), shell)
-	if err != nil {
-		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, err
-	}
-
-	// Return Default Kafka Version If Not Specified
-	if len(shell.Version) <= 0 {
-		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, nil
-	}
-
-	// Attempt To Parse The Version String Into A Sarama.KafkaVersion
-	kafkaVersion, err := sarama.ParseKafkaVersion(shell.Version)
-	if err != nil {
-		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, err
-	}
-
-	// Remove The Version From The Sarama Config YAML String
-	regex, err := regexp.Compile("\\s*Version:\\s*" + shell.Version + "\\s*")
-	if err != nil {
-		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, err
-	} else {
-		updatedSaramaConfigYamlBytes := regex.ReplaceAll([]byte(saramaConfigYamlString), []byte{})
-		return string(updatedSaramaConfigYamlBytes), kafkaVersion, nil
-	}
-}
-
-/* Extract (Parse & Remove) TLS.Config Level RootPEMs From Specified Sarama Confirm YAML String
-
-The Sarama.Config struct contains Net.TLS.Config which is a *tls.Config which cannot be parsed.
-due to it being from another package and containing lots of func()s.  We do need the ability
-however to provide Root Certificates in order to avoid having to disable verification via the
-InsecureSkipVerify field.  Therefore, we support a custom 'RootPEMs' field which is an array
-of strings containing the PEM file content.  This function will "extract" that content out
-of the YAML string and return as a populated *x509.CertPool which can be assigned to the
-Sarama.Config.Net.TLS.Config.RootCAs field.  In the case where the user has NOT specified
-any PEM files we will return nil.
-
-In order for the PEM file content to pass cleanly from the YAML string, which is itself
-already a YAML string inside the K8S ConfigMap, it must be formatted and spaced correctly.
-The following shows an example of the expected usage...
-
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: config-kafka
-  namespace: knative-eventing
-data:
-  sarama: |
-    Admin:
-      Timeout: 10000000000
-    Net:
-      KeepAlive: 30000000000
-      TLS:
-        Enable: true
-        Config:
-          RootCaPems:
-          - |-
-            -----BEGIN CERTIFICATE-----
-            MIIGBDCCA+ygAwIBAgIJAKi1aEV58cQ1MA0GCSqGSIb3DQEBCwUAMIGOMQswCQYD
-            ...
-            2wk9rLRZaQnhspt6MhlmU0qkaEZpYND3emR2XZ07m51jXqDUgTjXYCSggImUsARs
-            NAehp9bMeco=
-            -----END CERTIFICATE-----
-      SASL:
-        Enable: true
-
-...where you should make sure to use the YAML string syntax of "|-" in order to
-prevent trailing linefeed. The indentation of the PEM content is also important
-and must be aligned as shown.
-*/
-func extractRootCerts(saramaConfigYamlString string) (string, *x509.CertPool, error) {
-
-	// Define Inline Struct To Marshall The TLS Config 'RootPEMs' Into
-	type tlsConfigShell struct {
-		Net struct {
-			TLS struct {
-				Config struct {
-					RootPEMs []string
-				}
-			}
-		}
-	}
-
-	// Unmarshal The TLS Config Into The Shell
-	shell := &tlsConfigShell{}
-	err := yaml.Unmarshal([]byte(saramaConfigYamlString), shell)
-	if err != nil {
-		return saramaConfigYamlString, nil, err
-	}
-
-	// Convenience Variable For The RootPEMs
-	rootPEMs := shell.Net.TLS.Config.RootPEMs
-
-	// Exit Early If No RootCert PEMs
-	if rootPEMs == nil || len(rootPEMs) < 1 {
-		return saramaConfigYamlString, nil, nil
-	}
-
-	// Create A New CertPool To Contain The Root PEMs
-	certPool := x509.NewCertPool()
-
-	// Populate The CertPool With The PEMs
-	for _, rootPEM := range rootPEMs {
-		ok := certPool.AppendCertsFromPEM([]byte(rootPEM))
-		if !ok {
-			return saramaConfigYamlString, nil, fmt.Errorf("failed to parse root certificate PEM: %s", rootPEM)
-		}
-	}
-
-	// Remove The RootPEMs From The Sarama YAML String (Multi-Line / Greedy To Collect All PEMs)
-	updatedSaramaConfigYamlBytes := regexRootPEMs.ReplaceAll([]byte(saramaConfigYamlString), []byte{})
-	return string(updatedSaramaConfigYamlBytes), certPool, nil
-}
-
 // ConfigEqual is a convenience function to determine if two given sarama.Config structs are identical aside
 // from unserializable fields (e.g. function pointers).  To ignore parts of the sarama.Config struct, pass
 // them in as the "ignore" parameter.
@@ -235,60 +81,6 @@ func ConfigEqual(config1, config2 *sarama.Config, ignore ...interface{}) bool {
 	return cmp.Equal(config1, config2, ignoredTypes, ignoredUnexported)
 }
 
-// Extract The Sarama-Specific Settings From A ConfigMap And Merge Them With Existing Settings
-// If config Is nil, A New sarama.Config Struct Will Be Created With Default Values
-func MergeSaramaSettings(config *sarama.Config, configMap *corev1.ConfigMap) (*sarama.Config, error) {
-
-	// Validate The ConfigMap Data
-	if configMap.Data == nil {
-		return nil, fmt.Errorf("attempted to merge sarama settings with empty configmap")
-	}
-
-	// Merging To A Nil Config Requires Creating An Default One First
-	if config == nil {
-		// Start With Base Sarama Defaults
-		config = sarama.NewConfig()
-
-		// Use Our Default Minimum Version
-		config.Version = constants.ConfigKafkaVersionDefault
-
-		// Add Any Required Settings
-		UpdateSaramaConfig(config, config.ClientID, "", "")
-	}
-
-	// Merge The ConfigMap Settings Into The Provided Config
-	saramaSettingsYamlString := configMap.Data[testing.SaramaSettingsConfigKey]
-
-	// Extract (Remove) The KafkaVersion From The Sarama Config YAML
-	saramaSettingsYamlString, kafkaVersion, err := extractKafkaVersion(saramaSettingsYamlString)
-	if err != nil {
-		return nil, fmt.Errorf("failed to extract KafkaVersion from Sarama Config YAML: err=%s : config=%+v", err, saramaSettingsYamlString)
-	}
-
-	// Extract (Remove) Any TLS.Config RootCAs & Set In Sarama.Config
-	saramaSettingsYamlString, certPool, err := extractRootCerts(saramaSettingsYamlString)
-	if err != nil {
-		return nil, fmt.Errorf("failed to extract RootPEMs from Sarama Config YAML: err=%s : config=%+v", err, saramaSettingsYamlString)
-	}
-
-	// Unmarshall The Sarama Config Yaml Into The Provided Sarama.Config Object
-	err = yaml.Unmarshal([]byte(saramaSettingsYamlString), &config)
-	if err != nil {
-		return nil, fmt.Errorf("ConfigMap's sarama value could not be converted to a Sarama.Config struct: %s : %v", err, saramaSettingsYamlString)
-	}
-
-	// Override The Custom Parsed KafkaVersion
-	config.Version = kafkaVersion
-
-	// Override Any Custom Parsed TLS.Config.RootCAs
-	if certPool != nil && len(certPool.Subjects()) > 0 {
-		config.Net.TLS.Config = &tls.Config{RootCAs: certPool}
-	}
-
-	// Return Success
-	return config, nil
-}
-
 // Load The Sarama & EventingKafka Configuration From The ConfigMap
 // The Provided Context Must Have A Kubernetes Client Associated With It
 func LoadSettings(ctx context.Context) (*sarama.Config, *commonconfig.EventingKafkaConfig, error) {
@@ -306,8 +98,16 @@ func LoadSettings(ctx context.Context) (*sarama.Config, *commonconfig.EventingKa
 		return nil, nil, err
 	}
 
+	// Validate The ConfigMap Data
+	if configMap.Data == nil {
+		return nil, nil, fmt.Errorf("Attempted to merge sarama settings with empty configmap")
+	}
+
+	// Merge The ConfigMap Settings Into The Provided Config
+	saramaSettingsYamlString := configMap.Data[testing.SaramaSettingsConfigKey]
+
 	// Merge The Sarama Settings In The ConfigMap Into A New Base Sarama Config
-	saramaConfig, err := MergeSaramaSettings(nil, configMap)
+	saramaConfig, err := client.MergeSaramaSettings(nil, saramaSettingsYamlString)
 
 	return saramaConfig, eventingKafkaConfig, err
 }
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler.go b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
index 2b35ac81f..b00582ffb 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
@@ -32,6 +32,7 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	kafkaadmin "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin"
 	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
@@ -39,6 +40,7 @@ import (
 	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
 	kafkalisters "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/reconciler"
 )
@@ -241,8 +243,17 @@ func (r *Reconciler) configMapObserver(configMap *corev1.ConfigMap) {
 	// env.GetEnvironment is not necessary now.  If	those settings are needed in the future, the
 	// environment will also need to be re-parsed here.
 
+	// Validate The ConfigMap Data
+	if configMap.Data == nil {
+		r.logger.Fatal("Attempted to merge sarama settings with empty configmap")
+		return
+	}
+
+	// Merge The ConfigMap Settings Into The Provided Config
+	saramaSettingsYamlString := configMap.Data[testing.SaramaSettingsConfigKey]
+
 	// Load the Sarama settings from our configmap, ignoring the eventing-kafka result.
-	saramaConfig, err := kafkasarama.MergeSaramaSettings(nil, configMap)
+	saramaConfig, err := client.MergeSaramaSettings(nil, saramaSettingsYamlString)
 	if err != nil {
 		r.logger.Fatal("Failed To Load Eventing-Kafka Settings", zap.Error(err))
 	}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 7c05aeca9..f1ee8e5e5 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -28,6 +28,8 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer"
 	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/metrics"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/channel"
 )
@@ -272,7 +274,16 @@ func (d *DispatcherImpl) ConfigChanged(configMap *v1.ConfigMap) Dispatcher {
 	// Create A New Sarama Config
 	d.Logger.Debug("New ConfigMap Received", zap.String("configMap.Name", configMap.ObjectMeta.Name))
 
-	newConfig, err := kafkasarama.MergeSaramaSettings(nil, configMap)
+	// Validate The ConfigMap Data
+	if configMap.Data == nil {
+		d.Logger.Error("Attempted to merge sarama settings with empty configmap")
+		return nil
+	}
+
+	// Merge The ConfigMap Settings Into The Provided Config
+	saramaSettingsYamlString := configMap.Data[testing.SaramaSettingsConfigKey]
+
+	newConfig, err := client.MergeSaramaSettings(nil, saramaSettingsYamlString)
 	if err != nil {
 		d.Logger.Error("Unable to merge sarama settings", zap.Error(err))
 		return nil
@@ -282,7 +293,7 @@ func (d *DispatcherImpl) ConfigChanged(configMap *v1.ConfigMap) Dispatcher {
 	if d.SaramaConfig != nil {
 
 		// Some of the current config settings may not be overridden by the configmap (username, password, etc.)
-		kafkasarama.UpdateSaramaConfig(newConfig, d.SaramaConfig.ClientID, d.SaramaConfig.Net.SASL.User, d.SaramaConfig.Net.SASL.Password)
+		client.UpdateSaramaConfig(newConfig, d.SaramaConfig.ClientID, d.SaramaConfig.Net.SASL.User, d.SaramaConfig.Net.SASL.Password)
 
 		// Enable Sarama Logging If Specified In ConfigMap
 		if ekConfig, err := kafkasarama.LoadEventingKafkaSettings(configMap); err == nil && ekConfig != nil {
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index 5fd0e5d6a..1fa2b5780 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -21,6 +21,8 @@ import (
 	"errors"
 	"time"
 
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	"knative.dev/eventing-kafka/pkg/common/tracing"
 
 	"go.opencensus.io/trace"
@@ -199,8 +201,17 @@ func (p *Producer) ConfigChanged(configMap *v1.ConfigMap) *Producer {
 	// Create A New Sarama Config (Carrying Forward The Kafka Secret Auth)
 	p.logger.Debug("New ConfigMap Received", zap.String("configMap.Name", configMap.ObjectMeta.Name))
 
+	// Validate The ConfigMap Data
+	if configMap.Data == nil {
+		p.logger.Error("Attempted to merge sarama settings with empty configmap")
+		return nil
+	}
+
+	// Merge The ConfigMap Settings Into The Provided Config
+	saramaSettingsYamlString := configMap.Data[testing.SaramaSettingsConfigKey]
+
 	// Merge The Sarama Config From ConfigMap Into New Sarama Config
-	newConfig, err := kafkasarama.MergeSaramaSettings(nil, configMap)
+	newConfig, err := client.MergeSaramaSettings(nil, saramaSettingsYamlString)
 	if err != nil {
 		p.logger.Error("Unable to merge sarama settings", zap.Error(err))
 		return nil
@@ -210,7 +221,7 @@ func (p *Producer) ConfigChanged(configMap *v1.ConfigMap) *Producer {
 	if p.configuration != nil {
 
 		// Some of the current config settings may not be overridden by the configmap (username, password, etc.)
-		kafkasarama.UpdateSaramaConfig(newConfig, p.configuration.ClientID, p.configuration.Net.SASL.User, p.configuration.Net.SASL.Password)
+		client.UpdateSaramaConfig(newConfig, p.configuration.ClientID, p.configuration.Net.SASL.User, p.configuration.Net.SASL.Password)
 
 		// Enable Sarama Logging If Specified In ConfigMap
 		if ekConfig, err := kafkasarama.LoadEventingKafkaSettings(configMap); err == nil && ekConfig != nil {
diff --git a/pkg/common/client/config.go b/pkg/common/client/config.go
new file mode 100644
index 000000000..0f042dbba
--- /dev/null
+++ b/pkg/common/client/config.go
@@ -0,0 +1,89 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"crypto/tls"
+	"fmt"
+
+	"github.com/Shopify/sarama"
+	"github.com/ghodss/yaml"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
+)
+
+// Merge existing Sarama config with provided YAML string.
+// Values in YAML string will override the values in the config.
+// If config Is nil, A New sarama.Config Struct Will Be Created With Default Values
+func MergeSaramaSettings(config *sarama.Config, saramaSettingsYamlString string) (*sarama.Config, error) {
+	// Merging To A Nil Config Requires Creating An Default One First
+	if config == nil {
+		// Start With Base Sarama Defaults
+		config = sarama.NewConfig()
+
+		// Use Our Default Minimum Version
+		config.Version = constants.ConfigKafkaVersionDefault
+
+		// Add Any Required Settings
+		UpdateSaramaConfig(config, config.ClientID, "", "")
+	}
+
+	// Extract (Remove) The KafkaVersion From The Sarama Config YAML
+	saramaSettingsYamlString, kafkaVersion, err := extractKafkaVersion(saramaSettingsYamlString)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract KafkaVersion from Sarama Config YAML: err=%s : config=%+v", err, saramaSettingsYamlString)
+	}
+
+	// Extract (Remove) Any TLS.Config RootCAs & Set In Sarama.Config
+	saramaSettingsYamlString, certPool, err := extractRootCerts(saramaSettingsYamlString)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract RootPEMs from Sarama Config YAML: err=%s : config=%+v", err, saramaSettingsYamlString)
+	}
+
+	// Unmarshall The Sarama Config Yaml Into The Provided Sarama.Config Object
+	err = yaml.Unmarshal([]byte(saramaSettingsYamlString), &config)
+	if err != nil {
+		return nil, fmt.Errorf("ConfigMap's sarama value could not be converted to a Sarama.Config struct: %s : %v", err, saramaSettingsYamlString)
+	}
+
+	// Override The Custom Parsed KafkaVersion
+	config.Version = kafkaVersion
+
+	// Override Any Custom Parsed TLS.Config.RootCAs
+	if certPool != nil && len(certPool.Subjects()) > 0 {
+		config.Net.TLS.Config = &tls.Config{RootCAs: certPool}
+	}
+
+	// Return Success
+	return config, nil
+}
+
+// Utility Function For Configuring Common Settings For Admin/Producer/Consumer
+func UpdateSaramaConfig(config *sarama.Config, clientId string, username string, password string) {
+
+	// Set The ClientID For Logging
+	config.ClientID = clientId
+
+	// Set The SASL Username / Password
+	config.Net.SASL.User = username
+	config.Net.SASL.Password = password
+
+	// We Always Want To Know About Consumer Errors
+	config.Consumer.Return.Errors = true
+
+	// We Always Want Success Messages From Producer
+	config.Producer.Return.Successes = true
+}
diff --git a/pkg/common/client/sarama.go b/pkg/common/client/sarama.go
new file mode 100644
index 000000000..73eb62e8a
--- /dev/null
+++ b/pkg/common/client/sarama.go
@@ -0,0 +1,163 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"crypto/x509"
+	"fmt"
+	"regexp"
+
+	"github.com/Shopify/sarama"
+	"github.com/ghodss/yaml"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
+)
+
+// Regular Expression To Find All Certificates In Net.TLS.Config.RootPEMs Field
+var regexRootPEMs = regexp.MustCompile(`(?s)\s*RootPEMs:.*-----END CERTIFICATE-----`)
+
+//
+// Extract (Parse & Remove) Top Level Kafka Version From Specified Sarama Confirm YAML String
+//
+// The Sarama.Config struct contains a top-level 'Version' field of type sarama.KafkaVersion.
+// This type contains a package scoped [4]uint field called 'version' which we cannot unmarshal
+// the yaml into.  Therefore, we instead support the user providing a 'Version' string of the
+// format "major.minor.patch" (e.g. "2.3.0") which we will "extract" out of the YAML string
+// and return as an actual sarama.KafkaVersion.  In the case where the user has NOT specified
+// a Version string we will return a default version.
+//
+func extractKafkaVersion(saramaConfigYamlString string) (string, sarama.KafkaVersion, error) {
+
+	// Define Inline Struct To Marshall The Top Level Sarama Config "Kafka" Version Into
+	type saramaConfigShell struct {
+		Version string
+	}
+
+	// Unmarshal The Sarama Config Into The Shell
+	shell := &saramaConfigShell{}
+	err := yaml.Unmarshal([]byte(saramaConfigYamlString), shell)
+	if err != nil {
+		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, err
+	}
+
+	// Return Default Kafka Version If Not Specified
+	if len(shell.Version) <= 0 {
+		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, nil
+	}
+
+	// Attempt To Parse The Version String Into A Sarama.KafkaVersion
+	kafkaVersion, err := sarama.ParseKafkaVersion(shell.Version)
+	if err != nil {
+		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, err
+	}
+
+	// Remove The Version From The Sarama Config YAML String
+	regex, err := regexp.Compile("\\s*Version:\\s*" + shell.Version + "\\s*")
+	if err != nil {
+		return saramaConfigYamlString, constants.ConfigKafkaVersionDefault, err
+	} else {
+		updatedSaramaConfigYamlBytes := regex.ReplaceAll([]byte(saramaConfigYamlString), []byte{})
+		return string(updatedSaramaConfigYamlBytes), kafkaVersion, nil
+	}
+}
+
+/* Extract (Parse & Remove) TLS.Config Level RootPEMs From Specified Sarama Confirm YAML String
+
+The Sarama.Config struct contains Net.TLS.Config which is a *tls.Config which cannot be parsed.
+due to it being from another package and containing lots of func()s.  We do need the ability
+however to provide Root Certificates in order to avoid having to disable verification via the
+InsecureSkipVerify field.  Therefore, we support a custom 'RootPEMs' field which is an array
+of strings containing the PEM file content.  This function will "extract" that content out
+of the YAML string and return as a populated *x509.CertPool which can be assigned to the
+Sarama.Config.Net.TLS.Config.RootCAs field.  In the case where the user has NOT specified
+any PEM files we will return nil.
+
+In order for the PEM file content to pass cleanly from the YAML string, which is itself
+already a YAML string inside the K8S ConfigMap, it must be formatted and spaced correctly.
+The following shows an example of the expected usage...
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka
+  namespace: knative-eventing
+data:
+  sarama: |
+    Admin:
+      Timeout: 10000000000
+    Net:
+      KeepAlive: 30000000000
+      TLS:
+        Enable: true
+        Config:
+          RootCaPems:
+          - |-
+            -----BEGIN CERTIFICATE-----
+            MIIGBDCCA+ygAwIBAgIJAKi1aEV58cQ1MA0GCSqGSIb3DQEBCwUAMIGOMQswCQYD
+            ...
+            2wk9rLRZaQnhspt6MhlmU0qkaEZpYND3emR2XZ07m51jXqDUgTjXYCSggImUsARs
+            NAehp9bMeco=
+            -----END CERTIFICATE-----
+      SASL:
+        Enable: true
+
+...where you should make sure to use the YAML string syntax of "|-" in order to
+prevent trailing linefeed. The indentation of the PEM content is also important
+and must be aligned as shown.
+*/
+func extractRootCerts(saramaConfigYamlString string) (string, *x509.CertPool, error) {
+
+	// Define Inline Struct To Marshall The TLS Config 'RootPEMs' Into
+	type tlsConfigShell struct {
+		Net struct {
+			TLS struct {
+				Config struct {
+					RootPEMs []string
+				}
+			}
+		}
+	}
+
+	// Unmarshal The TLS Config Into The Shell
+	shell := &tlsConfigShell{}
+	err := yaml.Unmarshal([]byte(saramaConfigYamlString), shell)
+	if err != nil {
+		return saramaConfigYamlString, nil, err
+	}
+
+	// Convenience Variable For The RootPEMs
+	rootPEMs := shell.Net.TLS.Config.RootPEMs
+
+	// Exit Early If No RootCert PEMs
+	if rootPEMs == nil || len(rootPEMs) < 1 {
+		return saramaConfigYamlString, nil, nil
+	}
+
+	// Create A New CertPool To Contain The Root PEMs
+	certPool := x509.NewCertPool()
+
+	// Populate The CertPool With The PEMs
+	for _, rootPEM := range rootPEMs {
+		ok := certPool.AppendCertsFromPEM([]byte(rootPEM))
+		if !ok {
+			return saramaConfigYamlString, nil, fmt.Errorf("failed to parse root certificate PEM: %s", rootPEM)
+		}
+	}
+
+	// Remove The RootPEMs From The Sarama YAML String (Multi-Line / Greedy To Collect All PEMs)
+	updatedSaramaConfigYamlBytes := regexRootPEMs.ReplaceAll([]byte(saramaConfigYamlString), []byte{})
+	return string(updatedSaramaConfigYamlBytes), certPool, nil
+}
diff --git a/vendor/knative.dev/pkg/controller/controller.go b/vendor/knative.dev/pkg/controller/controller.go
index 941117c71..341e4ab2d 100644
--- a/vendor/knative.dev/pkg/controller/controller.go
+++ b/vendor/knative.dev/pkg/controller/controller.go
@@ -26,6 +26,7 @@ import (
 	"github.com/google/uuid"
 
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -265,9 +266,12 @@ func (c *Impl) EnqueueAfter(obj interface{}, after time.Duration) {
 // and enqueues that key in the slow lane.
 func (c *Impl) EnqueueSlowKey(key types.NamespacedName) {
 	c.workQueue.SlowLane().Add(key)
-	c.logger.With(zap.String(logkey.Key, key.String())).
-		Debugf("Adding to the slow queue %s (depth(total/slow): %d/%d)",
-			safeKey(key), c.workQueue.Len(), c.workQueue.SlowLane().Len())
+
+	if logger := c.logger.Desugar(); logger.Core().Enabled(zapcore.DebugLevel) {
+		logger.Debug(fmt.Sprintf("Adding to the slow queue %s (depth(total/slow): %d/%d)",
+			safeKey(key), c.workQueue.Len(), c.workQueue.SlowLane().Len()),
+			zap.String(logkey.Key, key.String()))
+	}
 }
 
 // EnqueueSlow extracts namespaced name from the object and enqueues it on the slow
@@ -393,8 +397,11 @@ func (c *Impl) EnqueueNamespaceOf(obj interface{}) {
 // EnqueueKey takes a namespace/name string and puts it onto the work queue.
 func (c *Impl) EnqueueKey(key types.NamespacedName) {
 	c.workQueue.Add(key)
-	c.logger.With(zap.String(logkey.Key, key.String())).
-		Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.workQueue.Len())
+
+	if logger := c.logger.Desugar(); logger.Core().Enabled(zapcore.DebugLevel) {
+		logger.Debug(fmt.Sprintf("Adding to queue %s (depth: %d)", safeKey(key), c.workQueue.Len()),
+			zap.String(logkey.Key, key.String()))
+	}
 }
 
 // MaybeEnqueueBucketKey takes a Bucket and namespace/name string and puts it onto
@@ -409,8 +416,11 @@ func (c *Impl) MaybeEnqueueBucketKey(bkt reconciler.Bucket, key types.Namespaced
 // the work queue after given delay.
 func (c *Impl) EnqueueKeyAfter(key types.NamespacedName, delay time.Duration) {
 	c.workQueue.AddAfter(key, delay)
-	c.logger.With(zap.String(logkey.Key, key.String())).
-		Debugf("Adding to queue %s (delay: %v, depth: %d)", safeKey(key), delay, c.workQueue.Len())
+
+	if logger := c.logger.Desugar(); logger.Core().Enabled(zapcore.DebugLevel) {
+		logger.Debug(fmt.Sprintf("Adding to queue %s (delay: %v, depth: %d)", safeKey(key), delay, c.workQueue.Len()),
+			zap.String(logkey.Key, key.String()))
+	}
 }
 
 // RunContext starts the controller's worker threads, the number of which is threadiness.
diff --git a/vendor/knative.dev/pkg/logging/config.go b/vendor/knative.dev/pkg/logging/config.go
index 1effe1b35..7e8636f9b 100644
--- a/vendor/knative.dev/pkg/logging/config.go
+++ b/vendor/knative.dev/pkg/logging/config.go
@@ -195,8 +195,8 @@ func UpdateLevelFromConfigMap(logger *zap.SugaredLogger, atomicLevel zap.AtomicL
 			case errors.Is(err, errEmptyLoggerConfig):
 				level = zap.NewAtomicLevel().Level()
 			case err != nil:
-				logger.With(zap.Error(err)).Errorf("Failed to parse logger configuration. "+
-					"Previous log level retained for %v", levelKey)
+				logger.Errorw("Failed to parse logger configuration. Previous log level retained for "+levelKey,
+					zap.Error(err))
 				return
 			default:
 				level = loggingCfg.Level.Level()
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 20ac87bed..b6f96e767 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1077,7 +1077,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.19.1-0.20201216081458-3e06c6b996dd
+# knative.dev/eventing v0.19.1-0.20201216183558-f3468d1c9954
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/util/crstatusevent
@@ -1211,7 +1211,7 @@ knative.dev/eventing/test/test_images/recordevents
 # knative.dev/hack v0.0.0-20201214230143-4ed1ecb8db24
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20201216014558-95b8793bd623
+# knative.dev/pkg v0.0.0-20201217014702-b0c121fb667f
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/pkg/channel/distributed/common/kafka/consumer/consumer_test.go b/pkg/channel/distributed/common/kafka/consumer/consumer_test.go
index f67bc22fa..50834c077 100644
--- a/pkg/channel/distributed/common/kafka/consumer/consumer_test.go
+++ b/pkg/channel/distributed/common/kafka/consumer/consumer_test.go
@@ -23,9 +23,9 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
-	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
 	kafkatesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/testing"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/client"
 )
 
 // Test Constants
@@ -65,7 +65,7 @@ func TestCreateConsumerGroup(t *testing.T) {
 
 	// Perform The Test
 	config := commontesting.GetDefaultSaramaConfig(t)
-	kafkasarama.UpdateSaramaConfig(config, ClientId, KafkaUsername, KafkaPassword)
+	client.UpdateSaramaConfig(config, ClientId, KafkaUsername, KafkaPassword)
 	consumerGroup, registry, err := CreateConsumerGroup(brokers, config, GroupId)
 
 	// Verify The Results
@@ -77,7 +77,7 @@ func TestCreateConsumerGroup(t *testing.T) {
 // Test that the UpdateSaramaConfig sets values as expected
 func TestUpdateConfig(t *testing.T) {
 	config := sarama.NewConfig()
-	kafkasarama.UpdateSaramaConfig(config, ClientId, KafkaUsername, KafkaPassword)
+	client.UpdateSaramaConfig(config, ClientId, KafkaUsername, KafkaPassword)
 	assert.Equal(t, ClientId, config.ClientID)
 	assert.Equal(t, KafkaUsername, config.Net.SASL.User)
 	assert.Equal(t, KafkaPassword, config.Net.SASL.Password)
diff --git a/pkg/channel/distributed/common/kafka/producer/producer_test.go b/pkg/channel/distributed/common/kafka/producer/producer_test.go
index 8d39d2f8b..7ba1980d5 100644
--- a/pkg/channel/distributed/common/kafka/producer/producer_test.go
+++ b/pkg/channel/distributed/common/kafka/producer/producer_test.go
@@ -23,8 +23,8 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
-	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/client"
 )
 
 // Test Constants
@@ -64,7 +64,7 @@ func performCreateSyncProducerTest(t *testing.T, username string, password strin
 
 	// Perform The Test
 	config := commontesting.GetDefaultSaramaConfig(t)
-	kafkasarama.UpdateSaramaConfig(config, ClientId, username, password)
+	client.UpdateSaramaConfig(config, ClientId, username, password)
 	producer, registry, err := CreateSyncProducer([]string{KafkaBrokers}, config)
 
 	// Verify The Results
@@ -77,7 +77,7 @@ func performCreateSyncProducerTest(t *testing.T, username string, password strin
 // Test that the UpdateSaramaConfig sets values as expected
 func TestUpdateConfig(t *testing.T) {
 	config := sarama.NewConfig()
-	kafkasarama.UpdateSaramaConfig(config, ClientId, KafkaUsername, KafkaPassword)
+	client.UpdateSaramaConfig(config, ClientId, KafkaUsername, KafkaPassword)
 	assert.Equal(t, ClientId, config.ClientID)
 	assert.Equal(t, KafkaUsername, config.Net.SASL.User)
 	assert.Equal(t, KafkaPassword, config.Net.SASL.Password)
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
index 6995c986c..41520bc63 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -20,9 +20,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"regexp"
 	"strconv"
-	"strings"
 	"testing"
 	"time"
 
@@ -32,6 +30,7 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 )
 
@@ -126,25 +125,6 @@ Consumer:
     Retention: 604800000000000
   Return:
     Errors: true
-`
-	EKDefaultSaramaConfigWithInsecureSkipVerify = `
-Net:
-  TLS:
-    Enable: true
-    Config:
-      InsecureSkipVerify: true
-  SASL:
-    Mechanism: PLAIN
-    Version: 1
-Metadata:
-  RefreshFrequency: 300000000000
-Consumer:
-  Offsets:
-    AutoCommit:
-        Interval: 5000000000
-    Retention: 604800000000000
-  Return:
-    Errors: true
 `
 )
 
@@ -169,26 +149,6 @@ func TestEnableSaramaLogging(t *testing.T) {
 	sarama.Logger.Print("TestMessage - Should Be Hidden")
 }
 
-// Test The UpdateSaramaConfig() Functionality
-func TestUpdateSaramaConfig(t *testing.T) {
-
-	// Test Data
-	clientId := "TestClientId"
-	username := "TestUsername"
-	password := "TestPassword"
-
-	// Perform The Test
-	config := sarama.NewConfig()
-	UpdateSaramaConfig(config, clientId, username, password)
-
-	// Verify The Results
-	assert.Equal(t, clientId, config.ClientID)
-	assert.Equal(t, username, config.Net.SASL.User)
-	assert.Equal(t, password, config.Net.SASL.Password)
-	assert.Equal(t, sarama.V1_0_0_0, config.Version)
-	assert.Nil(t, config.Net.TLS.Config)
-}
-
 // This test is specifically to validate that our default settings (used in 200-eventing-kafka-configmap.yaml)
 // are valid.  If the defaults in the file change, change this test to match for verification purposes.
 func TestLoadDefaultSaramaSettings(t *testing.T) {
@@ -226,66 +186,6 @@ func TestLoadDefaultSaramaSettings(t *testing.T) {
 	assert.Equal(t, "azure", configuration.Kafka.AdminType)
 }
 
-// Verify that the JSON fragment can be loaded into a sarama.Config struct
-func TestMergeSaramaSettings(t *testing.T) {
-	// Setup Environment
-	commontesting.SetTestEnvironment(t)
-
-	// Get a default Sarama config for verification that we don't overwrite settings when we merge
-	defaultConfig := sarama.NewConfig()
-
-	// Verify a few settings in different parts of two separate sarama.Config structures
-	// Since it's a simple JSON merge we don't need to test every possible value.
-	config, err := MergeSaramaSettings(nil, commontesting.GetTestSaramaConfigMap(commontesting.OldSaramaConfig, commontesting.TestEKConfig))
-	assert.Nil(t, err)
-	assert.NotNil(t, config)
-	assert.Equal(t, commontesting.OldUsername, config.Net.SASL.User)
-	assert.Equal(t, defaultConfig.Producer.Timeout, config.Producer.Timeout)
-	assert.Equal(t, defaultConfig.Consumer.MaxProcessingTime, config.Consumer.MaxProcessingTime)
-
-	config, err = MergeSaramaSettings(config, commontesting.GetTestSaramaConfigMap(commontesting.NewSaramaConfig, commontesting.TestEKConfig))
-	assert.Nil(t, err)
-	assert.NotNil(t, config)
-	assert.Equal(t, sarama.V2_3_0_0, config.Version)
-	assert.Equal(t, commontesting.NewUsername, config.Net.SASL.User)
-	assert.Equal(t, defaultConfig.Producer.Timeout, config.Producer.Timeout)
-	assert.Equal(t, defaultConfig.Consumer.MaxProcessingTime, config.Consumer.MaxProcessingTime)
-
-	// Verify error when no Data section is provided
-	configEmpty := commontesting.GetTestSaramaConfigMap(commontesting.NewSaramaConfig, commontesting.TestEKConfig)
-	configEmpty.Data = nil
-	config, err = MergeSaramaSettings(config, configEmpty)
-	assert.NotNil(t, err)
-
-	// Verify error when an invalid Version is provided
-	configMap := commontesting.GetTestSaramaConfigMap(commontesting.NewSaramaConfig, commontesting.TestEKConfig)
-	regexVersion := regexp.MustCompile(`Version:\s*\d*\.[\d.]*`) // Must have at least one period or it will match the "Version: 1" in Net.SASL
-	configMap.Data[commontesting.SaramaSettingsConfigKey] =
-		regexVersion.ReplaceAllString(configMap.Data[commontesting.SaramaSettingsConfigKey], "Version: INVALID")
-	config, err = MergeSaramaSettings(config, configMap)
-	assert.NotNil(t, err)
-
-	// Verify error when an invalid RootPEMs is provided
-	configMap = commontesting.GetTestSaramaConfigMap(EKDefaultSaramaConfigWithRootCert, commontesting.TestEKConfig)
-	regexPEMs := regexp.MustCompile(`-----BEGIN CERTIFICATE-----`)
-	configMap.Data[commontesting.SaramaSettingsConfigKey] =
-		regexPEMs.ReplaceAllString(configMap.Data[commontesting.SaramaSettingsConfigKey], "INVALID CERT DATA")
-	config, err = MergeSaramaSettings(config, configMap)
-	assert.NotNil(t, err)
-
-	// Verify that the RootPEMs section is merged properly
-	configMap = commontesting.GetTestSaramaConfigMap(EKDefaultSaramaConfigWithRootCert, commontesting.TestEKConfig)
-	config, err = MergeSaramaSettings(config, configMap)
-	assert.Nil(t, err)
-	assert.NotNil(t, config.Net.TLS.Config.RootCAs)
-
-	// Verify that the InsecureSkipVerify flag can be set properly
-	configMap = commontesting.GetTestSaramaConfigMap(EKDefaultSaramaConfigWithInsecureSkipVerify, commontesting.TestEKConfig)
-	config, err = MergeSaramaSettings(config, configMap)
-	assert.Nil(t, err)
-	assert.True(t, config.Net.TLS.Config.InsecureSkipVerify)
-}
-
 // Verify that comparisons of sarama config structs function as expected
 func TestSaramaConfigEqual(t *testing.T) {
 	config1 := sarama.NewConfig()
@@ -340,9 +240,9 @@ func TestSaramaConfigEqual(t *testing.T) {
 	assert.True(t, ConfigEqual(config1, config2))
 
 	// Test config with TLS struct
-	config1, err := MergeSaramaSettings(nil, commontesting.GetTestSaramaConfigMap(EKDefaultSaramaConfigWithRootCert, commontesting.TestEKConfig))
+	config1, err := client.MergeSaramaSettings(nil, EKDefaultSaramaConfigWithRootCert)
 	assert.Nil(t, err)
-	config2, err = MergeSaramaSettings(nil, commontesting.GetTestSaramaConfigMap(EKDefaultSaramaConfigWithRootCert, commontesting.TestEKConfig))
+	config2, err = client.MergeSaramaSettings(nil, EKDefaultSaramaConfigWithRootCert)
 	assert.Nil(t, err)
 	assert.True(t, ConfigEqual(config1, config2))
 }
@@ -431,34 +331,3 @@ func getTestSaramaContext(t *testing.T, saramaConfig string, eventingKafkaConfig
 	assert.NotNil(t, ctx)
 	return ctx
 }
-
-// Test The ExtractRoots() Functionality
-func TestExtractRootCerts(t *testing.T) {
-
-	// The Sarama Config YAML String To Test
-	beforeSaramaConfigYaml := EKDefaultSaramaConfigWithRootCert
-
-	// Perform The Test (Extract The RootCert)
-	afterSaramaConfigYaml, certPool, err := extractRootCerts(beforeSaramaConfigYaml)
-
-	// Verify The RootCert Was Extracted Successfully & Returned In CertPool
-	assert.NotNil(t, afterSaramaConfigYaml)
-	assert.NotEqual(t, beforeSaramaConfigYaml, afterSaramaConfigYaml)
-	assert.False(t, strings.Contains(afterSaramaConfigYaml, "RootPEMs"))
-	assert.False(t, strings.Contains(afterSaramaConfigYaml, "-----BEGIN CERTIFICATE-----"))
-	assert.False(t, strings.Contains(afterSaramaConfigYaml, "-----END CERTIFICATE-----"))
-	assert.NotNil(t, certPool)
-	assert.Nil(t, err)
-	subjects := certPool.Subjects()
-	assert.NotNil(t, subjects)
-	assert.Len(t, subjects, 1)
-
-	// Attempt To Extract Again (Now That There Arent' Any RootPEMs)
-	finalSaramaConfigYaml, certPool, err := extractRootCerts(afterSaramaConfigYaml)
-
-	// Verify The YAML String Is Unchanged And CertPool Is Nil
-	assert.NotNil(t, finalSaramaConfigYaml)
-	assert.Equal(t, afterSaramaConfigYaml, finalSaramaConfigYaml)
-	assert.Nil(t, certPool)
-	assert.Nil(t, err)
-}
diff --git a/pkg/common/client/config_test.go b/pkg/common/client/config_test.go
new file mode 100644
index 000000000..f78dd638c
--- /dev/null
+++ b/pkg/common/client/config_test.go
@@ -0,0 +1,114 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"regexp"
+	"strings"
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+)
+
+const (
+	EKDefaultSaramaConfigWithInsecureSkipVerify = `
+Net:
+  TLS:
+    Enable: true
+    Config:
+      InsecureSkipVerify: true
+  SASL:
+    Mechanism: PLAIN
+    Version: 1
+Metadata:
+  RefreshFrequency: 300000000000
+Consumer:
+  Offsets:
+    AutoCommit:
+        Interval: 5000000000
+    Retention: 604800000000000
+  Return:
+    Errors: true
+`
+)
+
+// Verify that the JSON fragment can be loaded into a sarama.Config struct
+func TestMergeSaramaSettings(t *testing.T) {
+	// Setup Environment
+	commontesting.SetTestEnvironment(t)
+
+	// Get a default Sarama config for verification that we don't overwrite settings when we merge
+	defaultConfig := sarama.NewConfig()
+
+	// Verify a few settings in different parts of two separate sarama.Config structures
+	// Since it's a simple JSON merge we don't need to test every possible value.
+	config, err := MergeSaramaSettings(nil, commontesting.OldSaramaConfig)
+	assert.Nil(t, err)
+	assert.NotNil(t, config)
+	assert.Equal(t, commontesting.OldUsername, config.Net.SASL.User)
+	assert.Equal(t, defaultConfig.Producer.Timeout, config.Producer.Timeout)
+	assert.Equal(t, defaultConfig.Consumer.MaxProcessingTime, config.Consumer.MaxProcessingTime)
+
+	config, err = MergeSaramaSettings(config, commontesting.NewSaramaConfig)
+	assert.Nil(t, err)
+	assert.NotNil(t, config)
+	assert.Equal(t, sarama.V2_3_0_0, config.Version)
+	assert.Equal(t, commontesting.NewUsername, config.Net.SASL.User)
+	assert.Equal(t, defaultConfig.Producer.Timeout, config.Producer.Timeout)
+	assert.Equal(t, defaultConfig.Consumer.MaxProcessingTime, config.Consumer.MaxProcessingTime)
+
+	// Verify error when an invalid Version is provided
+	regexVersion := regexp.MustCompile(`Version:\s*\d*\.[\d.]*`) // Must have at least one period or it will match the "Version: 1" in Net.SASL
+	config, err = MergeSaramaSettings(config, regexVersion.ReplaceAllString(commontesting.NewSaramaConfig, "Version: INVALID"))
+	assert.NotNil(t, err)
+
+	// Verify error when an invalid RootPEMs is provided
+	config, err = MergeSaramaSettings(config, strings.Replace(EKDefaultSaramaConfigWithRootCert, "-----BEGIN CERTIFICATE-----", "INVALID CERT DATA", -1))
+	assert.NotNil(t, err)
+
+	// Verify that the RootPEMs section is merged properly
+	config, err = MergeSaramaSettings(config, EKDefaultSaramaConfigWithRootCert)
+	assert.Nil(t, err)
+	assert.NotNil(t, config.Net.TLS.Config.RootCAs)
+
+	// Verify that the InsecureSkipVerify flag can be set properly
+	config, err = MergeSaramaSettings(config, EKDefaultSaramaConfigWithInsecureSkipVerify)
+	assert.Nil(t, err)
+	assert.True(t, config.Net.TLS.Config.InsecureSkipVerify)
+}
+
+// Test The UpdateSaramaConfig() Functionality
+func TestUpdateSaramaConfig(t *testing.T) {
+
+	// Test Data
+	clientId := "TestClientId"
+	username := "TestUsername"
+	password := "TestPassword"
+
+	// Perform The Test
+	config := sarama.NewConfig()
+	UpdateSaramaConfig(config, clientId, username, password)
+
+	// Verify The Results
+	assert.Equal(t, clientId, config.ClientID)
+	assert.Equal(t, username, config.Net.SASL.User)
+	assert.Equal(t, password, config.Net.SASL.Password)
+	assert.Equal(t, sarama.V1_0_0_0, config.Version)
+	assert.Nil(t, config.Net.TLS.Config)
+}
diff --git a/pkg/common/client/sarama_test.go b/pkg/common/client/sarama_test.go
new file mode 100644
index 000000000..38d7b58ee
--- /dev/null
+++ b/pkg/common/client/sarama_test.go
@@ -0,0 +1,111 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const EKDefaultSaramaConfigWithRootCert = `
+Net:
+  TLS:
+    Enable: true
+    Config:
+      RootPEMs: # Array of Root Certificate PEM Files As Strings (Mind indentation and use '|-' Syntax To Avoid Terminating \n)
+      - |-
+        -----BEGIN CERTIFICATE-----
+        MIIGBDCCA+ygAwIBAgIJAKi1aEV58cQ1MA0GCSqGSIb3DQEBCwUAMIGOMQswCQYD
+        VQQGEwJERTEbMBkGA1UECAwSQmFkZW4tV3VlcnR0ZW1iZXJnMREwDwYDVQQHDAhX
+        YWxsZG9yZjEPMA0GA1UECgwGU0FQIFNFMR8wHQYDVQQLDBZTQVAgQ1AgRGF0YSBN
+        YW5hZ2VtZW50MR0wGwYDVQQDDBRTQVAgQ1AgS2Fma2EgUm9vdCBDQTAeFw0xNzEy
+        MDQxMzUxMjZaFw0yMTAzMTgxMzUxMjZaMIGOMQswCQYDVQQGEwJERTEbMBkGA1UE
+        CAwSQmFkZW4tV3VlcnR0ZW1iZXJnMREwDwYDVQQHDAhXYWxsZG9yZjEPMA0GA1UE
+        CgwGU0FQIFNFMR8wHQYDVQQLDBZTQVAgQ1AgRGF0YSBNYW5hZ2VtZW50MR0wGwYD
+        VQQDDBRTQVAgQ1AgS2Fma2EgUm9vdCBDQTCCAiIwDQYJKoZIhvcNAQEBBQADggIP
+        ADCCAgoCggIBAKCx+et7E53Znvy+bFB/y4IDjubIEZOg+nmCYmID2RV/6PGtHXLY
+        DEwSue+JDwGXp4sLziFFHhoSjPx6OKLvwd1ww//FraDiGbeJY0BsnkpWVRbQiNyK
+        fxDY+YCLhYTujdtPZqcPcCII4QnQk1PoOrmgHuONGqgjIVTuSOeGx6eIUh8JC3TW
+        Z7EY0qKbnxCsVmyZudsO5Sh8AcDXNHAHJImoJ3uhWwU5YheCv24Jn0UcD/X843Jo
+        J6PhhoCmrLTZCVYeirv9jQqTiks0IhjQEAL6m2W6UCJArePzyjY+HOaY20Umo8Lf
+        CVjR0SfZric9g2+2XHkBex/73AMJbvyCvwER8oHwO9iGNeuHbkDdaicotQ5D7Nap
+        uXLgPFm3y/CkqiBXoiqCJxy+duM3itmLeW/PbEtNMnbS0mG64tZHd9THFAh3I+ug
+        w1+cQWzYO24EcdPQzaX8CpVJ8Au7aYc9QyyaayfTr4YxGYtMO0zay9tchEyChhtK
+        koHmyISz1kxuudItoRDNnRdbfUX1QeKnYWsUtfeK5MED2dpUPO+IVp7qomdy+F4T
+        KdQDvOlKBRFsngmyBbGeGB5wjXwTjuLfC0j6VIlfW0yMKhuePbqSPbVjGTFVefRo
+        rgODPaIre72GtXjcaVISlqagFQgOurRE5Z9OLpgCrMsLdOqVJ9LnSNTrAgMBAAGj
+        YzBhMB0GA1UdDgQWBBRkTG0qgjz9anjV94RGJ+GAApaf3DAfBgNVHSMEGDAWgBRk
+        TG0qgjz9anjV94RGJ+GAApaf3DAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB/wQE
+        AwIBhjANBgkqhkiG9w0BAQsFAAOCAgEAjL3wUM+Kgzbii2F76/qK2C1asFJkVQRd
+        CMiOhlZDEJYaBPzucF2vhOygkMMuw4SojkbzWGEdaRrc4IR6wVe0CezVeBrVRtAQ
+        DmCzdxO0xEZkWNMmMnzPBiB6k4l5Y9WiOGWiCrzLcMi8fiXr4pJoaUirUsvGf7xf
+        rwR6preFeLIZAgUesxy1RV2p9JHYm+iHiQskovkGt5Xr2sKJ+za3vtQ7Tf52rqAI
+        LPdhZXrMsqcza7yVfiJtS0orn3Su489bj6j+/MKjYjS6DvrSnw1VfzW1eA0U9nYt
+        vP8PVeWGsNxyg3YSwTaPi9cZ5lhGCoUSf2pq1g+VLvR1bIV++UL9wUHl4D7m5V4f
+        jqve5XlMMxYPk9l0YcA4nMF4CxpPsFqzx2MYfbWb1/RiR1BaHqgx7dFWJt980vHp
+        wM4tudQei+uUPYjLte09jKGLpZot0DGLIVJhT4RXnDV1VFmalRjJhJKBBIj7JPba
+        NKWCBaob148p5gwZ4dr4N/yaaUhesdYPJjZn+uvO29/pvv+u80nkEEWW2KYOCd44
+        SMTAhWkj5lx3X8xj40GSCxCMP+Jq2VLasoJSNminWVJuUaTk3veHsQ1mkoRDAbr1
+        2wk9rLRZaQnhspt6MhlmU0qkaEZpYND3emR2XZ07m51jXqDUgTjXYCSggImUsARs
+        NAehp9bMeco=
+        -----END CERTIFICATE-----
+  SASL:
+    Mechanism: PLAIN
+    Version: 1
+Metadata:
+  RefreshFrequency: 300000000000
+Consumer:
+  Offsets:
+    AutoCommit:
+        Interval: 5000000000
+    Retention: 604800000000000
+  Return:
+    Errors: true
+`
+
+// Test The ExtractRoots() Functionality
+func TestExtractRootCerts(t *testing.T) {
+
+	// The Sarama Config YAML String To Test
+	beforeSaramaConfigYaml := EKDefaultSaramaConfigWithRootCert
+
+	// Perform The Test (Extract The RootCert)
+	afterSaramaConfigYaml, certPool, err := extractRootCerts(beforeSaramaConfigYaml)
+
+	// Verify The RootCert Was Extracted Successfully & Returned In CertPool
+	assert.NotNil(t, afterSaramaConfigYaml)
+	assert.NotEqual(t, beforeSaramaConfigYaml, afterSaramaConfigYaml)
+	assert.False(t, strings.Contains(afterSaramaConfigYaml, "RootPEMs"))
+	assert.False(t, strings.Contains(afterSaramaConfigYaml, "-----BEGIN CERTIFICATE-----"))
+	assert.False(t, strings.Contains(afterSaramaConfigYaml, "-----END CERTIFICATE-----"))
+	assert.NotNil(t, certPool)
+	assert.Nil(t, err)
+	subjects := certPool.Subjects()
+	assert.NotNil(t, subjects)
+	assert.Len(t, subjects, 1)
+
+	// Attempt To Extract Again (Now That There Arent' Any RootPEMs)
+	finalSaramaConfigYaml, certPool, err := extractRootCerts(afterSaramaConfigYaml)
+
+	// Verify The YAML String Is Unchanged And CertPool Is Nil
+	assert.NotNil(t, finalSaramaConfigYaml)
+	assert.Equal(t, afterSaramaConfigYaml, finalSaramaConfigYaml)
+	assert.Nil(t, certPool)
+	assert.Nil(t, err)
+}