Repository: knative-sandbox/eventing-kafka
Pull request #384 (2021-02-13 03:49:30)

Expand number of kafkasource fields that are now mutable, add e2e tests

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes
- 🧽 Update or clean up current behavior


**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
The kafkasource now allows for mutable fields for:
- Topics 
- BootstrapServers 
- Sink
- Net
- CloudEventOverrides
Previously, after creation, all fields were immutable.
```


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/384#" title="Author self-approved">lberk</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)~~ [lberk]

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384?src=pr&el=h1) Report
> Merging [#384](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384?src=pr&el=desc) (e1796d6) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/5cab2571692609ab83d57d64cbbc5f1f5fe5484f?el=desc) (5cab257) will **increase** coverage by `0.04%`.
> The diff coverage is `64.28%`.

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

```diff
@@            Coverage Diff             @@
##           master     #384      +/-   ##
==========================================
+ Coverage   75.12%   75.17%   +0.04%     
==========================================
  Files         124      124              
  Lines        4792     4817      +25     
==========================================
+ Hits         3600     3621      +21     
- Misses        969      971       +2     
- Partials      223      225       +2     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/apis/sources/v1beta1/kafka\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384/diff?src=pr&el=tree#diff-cGtnL2FwaXMvc291cmNlcy92MWJldGExL2thZmthX3ZhbGlkYXRpb24uZ28=) | `69.69% <64.28%> (+12.55%)` | :arrow_up: |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `62.29% <0.00%> (-3.81%)` | :arrow_down: |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `70.37% <0.00%> (-0.56%)` | :arrow_down: |
| [pkg/common/scheduler/statefulset/autoscaler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvYXV0b3NjYWxlci5nbw==) | `76.59% <0.00%> (+2.78%)` | :arrow_up: |
| [pkg/common/scheduler/statefulset/snapshot.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvc25hcHNob3QuZ28=) | `76.47% <0.00%> (+7.23%)` | :arrow_up: |
| [pkg/common/scheduler/statefulset/helpers.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/384/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvaGVscGVycy5nbw==) | `60.00% <0.00%> (+40.00%)` | :arrow_up: |

------

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

knative-prow-robot: @lberk: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | e1796d62fff91c86e236024730968696cb217ea4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/384/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1360435914643345408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/384). [Your PR dashboard](https://gubernator.knative.dev/pr/lberk).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @lberk: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | e1796d62fff91c86e236024730968696cb217ea4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/384/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1360435914643345408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | e1796d62fff91c86e236024730968696cb217ea4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/384/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1360435914576236544) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/384). [Your PR dashboard](https://gubernator.knative.dev/pr/lberk).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index 4e0947a9b..be7ce9a87 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -100,7 +100,7 @@ jobs:
 
       - name: Install Tools
         env:
-          WOKE_VERSION: v0.4.1
+          WOKE_VERSION: v0.5.0
         run: |
           TEMP_PATH="$(mktemp -d)"
           cd $TEMP_PATH
diff --git a/go.mod b/go.mod
index 1805633ff..3c8dfe94e 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,7 @@ require (
 	k8s.io/apimachinery v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
-	knative.dev/eventing v0.20.1-0.20210211204829-1c771ce4058e
+	knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
-	knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe
+	knative.dev/pkg v0.0.0-20210216013737-584933f8280b
 )
diff --git a/go.sum b/go.sum
index 55b2ad9de..c71c8155a 100644
--- a/go.sum
+++ b/go.sum
@@ -1244,15 +1244,17 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6 h1:+WnxoVtG8TMiudHBSEtrVL
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K8Hf8whTseBgJcg=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.20.1-0.20210211204829-1c771ce4058e h1:04sCGKSZOTOGsSVHW+GvwWrlOnOTk79Q1yfX3YZsvWk=
-knative.dev/eventing v0.20.1-0.20210211204829-1c771ce4058e/go.mod h1:7UBdmxOatDhmTgMMsD016+B82VRLbdiODDixN8TZlrA=
+knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf h1:kjfXZLChfTzeXRDaXKmNO9EsnuVokkgzxPgikOJBfWM=
+knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf/go.mod h1:lxvfkunFc5wCcvlVqu4eJ3wbWqY4bTpQEDhNOMWD8+c=
 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-20210208175252-a02dcff9ee26 h1:PkJB1M77wYBI1UF5vp8y4fRnYQDu1IkRWswp6CgRRSo=
 knative.dev/pkg v0.0.0-20210208175252-a02dcff9ee26/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe h1:DH5UryA0fIXdNm9BYnqYo7X1IU8LlkVv8dd2ghkm8m8=
-knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/reconciler-test v0.0.0-20210210224717-0428cb00fe76/go.mod h1:hIOndGzuU1mxSJ8IF0RqPL/rAoazZZ0NjuJoMwKeNBM=
+knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f h1:pz+UDOTsCIClHXGM/SgCzkctLxgi3+P0C26x6Ujlqho=
+knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f/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-20210214052044-2a6803e31481/go.mod h1:hIOndGzuU1mxSJ8IF0RqPL/rAoazZZ0NjuJoMwKeNBM=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 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/apis/sources/v1beta1/kafka_validation.go b/pkg/apis/sources/v1beta1/kafka_validation.go
index 573fd321b..c68a20e7c 100644
--- a/pkg/apis/sources/v1beta1/kafka_validation.go
+++ b/pkg/apis/sources/v1beta1/kafka_validation.go
@@ -26,25 +26,58 @@ import (
 )
 
 // Validate ensures KafkaSource is properly configured.
-func (r *KafkaSource) Validate(ctx context.Context) *apis.FieldError {
+func (ks *KafkaSource) Validate(ctx context.Context) *apis.FieldError {
+	errs := ks.Spec.Validate(ctx).ViaField("spec")
 	if apis.IsInUpdate(ctx) {
 		original := apis.GetBaseline(ctx).(*KafkaSource)
-		// Ignore diffs in bootstrapServers and topics
-		ignoreArgs := cmpopts.IgnoreFields(KafkaSourceSpec{}, "Topics", "BootstrapServers")
-		if diff, err := kmp.ShortDiff(original.Spec, r.Spec, ignoreArgs); err != nil {
-			return &apis.FieldError{
-				Message: "Failed to diff KafkaSource",
-				Paths:   []string{"spec"},
-				Details: err.Error(),
-			}
-		} else if diff != "" {
-			return &apis.FieldError{
+		errs = errs.Also(ks.CheckImmutableFields(ctx, original))
+	}
+	return errs
+}
+
+func (kss *KafkaSourceSpec) Validate(ctx context.Context) *apis.FieldError {
+	var errs *apis.FieldError
+
+	// Validate sink
+	errs = errs.Also(kss.Sink.Validate(ctx).ViaField("sink"))
+
+	// We want to ensure we have a bootstrapServer & Topics for creation
+	if apis.IsInCreate(ctx) {
+		if len(kss.Topics) <= 0 {
+			errs = errs.Also(apis.ErrMissingField("Topics"))
+		}
+		if len(kss.BootstrapServers) <= 0 {
+			errs = errs.Also(apis.ErrMissingField("bootstrapServer"))
+		}
+	}
+
+	return errs
+}
+
+func (ks *KafkaSource) CheckImmutableFields(ctx context.Context, original *KafkaSource) *apis.FieldError {
+	if original == nil {
+		return nil
+	}
+	var errs *apis.FieldError
+
+	// Ignore diffs in bootstrapServers and topics
+	ignoreArgs := cmpopts.IgnoreFields(KafkaSourceSpec{}, "Topics", "BootstrapServers", "Sink", "Net", "CloudEventOverrides")
+	if diff, err := kmp.ShortDiff(original.Spec, ks.Spec, ignoreArgs); err != nil {
+		errs = errs.Also(&apis.FieldError{
+			Message: "Failed to diff KafkaSource",
+			Paths:   []string{"spec"},
+			Details: err.Error(),
+		})
+		return errs
+	} else if diff != "" {
+		errs = errs.Also(
+			&apis.FieldError{
 				Message: "Immutable fields changed (-old +new)",
 				Paths:   []string{"spec"},
 				Details: diff,
-			}
-		}
+			})
+		return errs
 	}
 
-	return nil
+	return errs
 }
diff --git a/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
index 8b808c988..353811c26 100644
--- a/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
+++ b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
@@ -145,7 +145,9 @@ func (w *WatcherImpl) Terminate() {
 
 	w.watchers = nil
 	w.cachedConsumerGroups = nil
-	w.done <- struct{}{}
+	if w.done != nil {
+		w.done <- struct{}{}
+	}
 }
 
 // TODO explore returning a channel instead of a taking callback
diff --git a/pkg/channel/consolidated/reconciler/controller/controller.go b/pkg/channel/consolidated/reconciler/controller/controller.go
index c19ee14a8..98cc260e0 100644
--- a/pkg/channel/consolidated/reconciler/controller/controller.go
+++ b/pkg/channel/consolidated/reconciler/controller/controller.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/kelseyhightower/envconfig"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"go.uber.org/zap"
 
@@ -87,12 +88,12 @@ func NewController(
 	impl := kafkaChannelReconciler.NewImpl(ctx, r)
 
 	// Get and Watch the Kakfa config map and dynamically update Kafka configuration.
-	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(system.Namespace()).Get(ctx, "config-kafka", metav1.GetOptions{}); err == nil {
-		cmw.Watch("config-kafka", func(configMap *v1.ConfigMap) {
+	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(system.Namespace()).Get(ctx, constants.SettingsConfigMapName, metav1.GetOptions{}); err == nil {
+		cmw.Watch(constants.SettingsConfigMapName, func(configMap *v1.ConfigMap) {
 			r.updateKafkaConfig(ctx, configMap)
 		})
 	} else if !apierrors.IsNotFound(err) {
-		logging.FromContext(ctx).With(zap.Error(err)).Fatal("Error reading ConfigMap 'config-kafka'")
+		logging.FromContext(ctx).With(zap.Error(err)).Fatalf("Error reading ConfigMap '%s'", constants.SettingsConfigMapName)
 	}
 
 	logging.FromContext(ctx).Info("Setting up event handlers")
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index 93ff5206d..a48c6d1bb 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -18,13 +18,13 @@ package controller
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strings"
 	"time"
 
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/kafka"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -50,7 +50,6 @@ import (
 	kafkaScheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
 	kafkaChannelReconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
 	listers "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
-	"knative.dev/eventing-kafka/pkg/source"
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/apis/eventing"
 	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
@@ -159,7 +158,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 
 	if r.kafkaConfig == nil {
 		if r.kafkaConfigError == nil {
-			r.kafkaConfigError = errors.New("The config map 'config-kafka' does not exist")
+			r.kafkaConfigError = fmt.Errorf("the config map '%s' does not exist", constants.SettingsConfigMapName)
 		}
 		kc.Status.MarkConfigFailed("MissingConfiguration", "%v", r.kafkaConfigError)
 		return r.kafkaConfigError
@@ -532,7 +531,7 @@ func (r *Reconciler) createClient(ctx context.Context) (sarama.ClusterAdmin, err
 	kafkaClusterAdmin := r.kafkaClusterAdmin
 	if kafkaClusterAdmin == nil {
 		var err error
-		kafkaClusterAdmin, err = source.MakeAdminClient(ctx, controllerAgentName, r.kafkaAuthConfig, r.kafkaConfig)
+		kafkaClusterAdmin, err = client.MakeAdminClient(ctx, controllerAgentName, r.kafkaAuthConfig, r.kafkaConfig.SaramaSettingsYamlString, r.kafkaConfig.Brokers)
 		if err != nil {
 			return nil, err
 		}
@@ -592,7 +591,7 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 	r.kafkaConfig = kafkaConfig
 	r.kafkaConfigError = err
 	ac, err := kafka.NewAdminClient(ctx, func() (sarama.ClusterAdmin, error) {
-		return source.MakeAdminClient(ctx, controllerAgentName, r.kafkaAuthConfig, kafkaConfig)
+		return client.MakeAdminClient(ctx, controllerAgentName, r.kafkaAuthConfig, kafkaConfig.SaramaSettingsYamlString, kafkaConfig.Brokers)
 	})
 
 	if err != nil {
@@ -601,7 +600,9 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 	}
 
 	if r.consumerGroupWatcher != nil {
+		logger.Info("terminating consumer group watcher")
 		r.consumerGroupWatcher.Terminate()
+		logger.Info("terminated consumer group watcher")
 	}
 
 	r.consumerGroupWatcher = NewConsumerGroupWatcher(ctx, ac, pollInterval)
@@ -617,6 +618,8 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, kc *v1beta1.KafkaChannel)
 			return err
 		}
 	}
-	r.consumerGroupWatcher.Forget(string(kc.ObjectMeta.UID))
+	if r.consumerGroupWatcher != nil {
+		r.consumerGroupWatcher.Forget(string(kc.ObjectMeta.UID))
+	}
 	return newReconciledNormal(kc.Namespace, kc.Name) //ok to remove finalizer
 }
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
index 63fdfba7e..ac34d20be 100644
--- a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
+++ b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
@@ -20,6 +20,7 @@ import (
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/system"
 )
 
@@ -77,7 +78,7 @@ func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
 							VolumeMounts: []corev1.VolumeMount{
 								{
 									Name:      "config-kafka",
-									MountPath: "/etc/config-kafka",
+									MountPath: constants.SettingsConfigMapMountPath,
 								},
 							},
 						},
@@ -88,7 +89,7 @@ func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
 							VolumeSource: corev1.VolumeSource{
 								ConfigMap: &corev1.ConfigMapVolumeSource{
 									LocalObjectReference: corev1.LocalObjectReference{
-										Name: "config-kafka",
+										Name: constants.SettingsConfigMapName,
 									},
 								},
 							},
diff --git a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index e1e01a38d..87e0608ef 100644
--- a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"knative.dev/eventing/pkg/apis/eventing"
 	"knative.dev/eventing/pkg/channel/fanout"
@@ -76,7 +77,7 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		logger.Fatalw("unable to setup tracing", zap.Error(err))
 	}
 
-	configMap, err := configmap.Load("/etc/config-kafka")
+	configMap, err := configmap.Load(constants.SettingsConfigMapMountPath)
 	if err != nil {
 		logger.Fatalw("error loading configuration", zap.Error(err))
 	}
diff --git a/pkg/common/client/admin_client.go b/pkg/common/client/admin_client.go
new file mode 100644
index 000000000..779cc9a58
--- /dev/null
+++ b/pkg/common/client/admin_client.go
@@ -0,0 +1,22 @@
+package client
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Shopify/sarama"
+)
+
+func MakeAdminClient(ctx context.Context, clientID string, kafkaAuthCfg *KafkaAuthConfig, saramaSettingsYamlString string, brokers []string) (sarama.ClusterAdmin, error) {
+	saramaConf, err := NewConfigBuilder().
+		WithDefaults().
+		WithAuth(kafkaAuthCfg).
+		WithClientId(clientID).
+		FromYaml(saramaSettingsYamlString).
+		Build(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("error creating admin client Sarama config: %w", err)
+	}
+
+	return sarama.NewClusterAdmin(brokers, saramaConf)
+}
diff --git a/pkg/common/constants/constants.go b/pkg/common/constants/constants.go
index 1b4272980..afb90ceae 100644
--- a/pkg/common/constants/constants.go
+++ b/pkg/common/constants/constants.go
@@ -23,8 +23,14 @@ const (
 	DefaultReplicationFactor = 1
 
 	// The name of the configmap used to hold eventing-kafka settings
-	SettingsConfigMapName          = "config-kafka"
+	SettingsConfigMapName = "config-kafka"
+
+	// Mount path of the configmap used to hold eventing-kafka settings
+	SettingsConfigMapMountPath = "/etc/" + SettingsConfigMapName
+
+	// Config key of the config in the configmap used to hold eventing-kafka settings
 	EventingKafkaSettingsConfigKey = "eventing-kafka"
+
 	// The name of the keys in the Data section of the eventing-kafka configmap that holds Sarama and Eventing-Kafka configuration YAML
 	SaramaSettingsConfigKey = "sarama"
 )
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index c9f2772fd..0a3c7b198 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -37,7 +37,7 @@ import (
 	"knative.dev/eventing/pkg/kncloudevents"
 
 	"knative.dev/eventing-kafka/pkg/common/consumer"
-	kafkasource "knative.dev/eventing-kafka/pkg/source"
+	"knative.dev/eventing-kafka/pkg/source/client"
 )
 
 const (
@@ -46,7 +46,7 @@ const (
 
 type AdapterConfig struct {
 	adapter.EnvConfig
-	kafkasource.KafkaEnvConfig
+	client.KafkaEnvConfig
 
 	Topics        []string `envconfig:"KAFKA_TOPICS" required:"true"`
 	ConsumerGroup string   `envconfig:"KAFKA_CONSUMER_GROUP" required:"true"`
@@ -93,7 +93,7 @@ func (a *Adapter) Start(ctx context.Context) error {
 	)
 
 	// init consumer group
-	addrs, config, err := kafkasource.NewConfigWithEnv(context.Background(), &a.config.KafkaEnvConfig)
+	addrs, config, err := client.NewConfigWithEnv(context.Background(), &a.config.KafkaEnvConfig)
 	if err != nil {
 		return fmt.Errorf("failed to create the config: %w", err)
 	}
diff --git a/pkg/source/client.go b/pkg/source/client/client.go
similarity index 76%
rename from pkg/source/client.go
rename to pkg/source/client/client.go
index 11d35da57..f07779430 100644
--- a/pkg/source/client.go
+++ b/pkg/source/client/client.go
@@ -14,17 +14,18 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package source
+package client
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
 
-	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/eventing-kafka/pkg/source/reconciler/source"
 )
 
 type AdapterSASL struct {
@@ -47,12 +48,15 @@ type AdapterNet struct {
 }
 
 type KafkaEnvConfig struct {
+	// KafkaConfigJson is the environment variable that's passed to adapter by the controller.
+	// It contains configuration from the Kafka configmap.
+	KafkaConfigJson  string   `envconfig:"K_KAFKA_CONFIG"`
 	BootstrapServers []string `envconfig:"KAFKA_BOOTSTRAP_SERVERS" required:"true"`
 	Net              AdapterNet
 }
 
 // NewConfig extracts the Kafka configuration from the environment.
-func NewConfig(ctx context.Context) ([]string, *sarama.Config, error) {
+func NewConfigFromEnv(ctx context.Context) ([]string, *sarama.Config, error) {
 	var env KafkaEnvConfig
 	if err := envconfig.Process("", &env); err != nil {
 		return nil, nil, err
@@ -81,11 +85,21 @@ func NewConfigWithEnv(ctx context.Context, env *KafkaEnvConfig) ([]string, *sara
 		}
 	}
 
-	cfg, err := client.NewConfigBuilder().
+	configBuilder := client.NewConfigBuilder().
 		WithDefaults().
-		WithAuth(kafkaAuthConfig).
-		WithVersion(&sarama.V2_0_0_0).
-		Build(ctx)
+		WithAuth(kafkaAuthConfig)
+
+	if env.KafkaConfigJson != "" {
+		kafkaCfg := &source.KafkaConfig{}
+		err := json.Unmarshal([]byte(env.KafkaConfigJson), kafkaCfg)
+		if err != nil {
+			return nil, nil, fmt.Errorf("error parsing Kafka config from environment: %w", err)
+		}
+
+		configBuilder = configBuilder.FromYaml(kafkaCfg.SaramaYamlString)
+	}
+
+	cfg, err := configBuilder.Build(ctx)
 	if err != nil {
 		return nil, nil, fmt.Errorf("error creating Sarama config: %w", err)
 	}
@@ -95,7 +109,7 @@ func NewConfigWithEnv(ctx context.Context, env *KafkaEnvConfig) ([]string, *sara
 
 // NewProducer is a helper method for constructing a client for producing kafka methods.
 func NewProducer(ctx context.Context) (sarama.Client, error) {
-	bs, cfg, err := NewConfig(ctx)
+	bs, cfg, err := NewConfigFromEnv(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -105,16 +119,3 @@ func NewProducer(ctx context.Context) (sarama.Client, error) {
 
 	return sarama.NewClient(bs, cfg)
 }
-
-func MakeAdminClient(ctx context.Context, clientID string, kafkaAuthCfg *client.KafkaAuthConfig, kafkaConfig *utils.KafkaConfig) (sarama.ClusterAdmin, error) {
-	saramaConf, err := client.NewConfigBuilder().
-		WithDefaults().
-		WithAuth(kafkaAuthCfg).
-		WithClientId(clientID).
-		Build(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("error creating admin client Sarama config: %w", err)
-	}
-
-	return sarama.NewClusterAdmin(kafkaConfig.Brokers, saramaConf)
-}
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index 7ff5c824a..ffc519265 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -35,8 +35,8 @@ import (
 
 	"knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/scheduler"
-	"knative.dev/eventing-kafka/pkg/source"
 	stadapter "knative.dev/eventing-kafka/pkg/source/adapter"
+	"knative.dev/eventing-kafka/pkg/source/client"
 )
 
 type AdapterConfig struct {
@@ -117,16 +117,16 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 			Component: "kafkasource",
 			Namespace: obj.Namespace,
 		},
-		KafkaEnvConfig: source.KafkaEnvConfig{
+		KafkaEnvConfig: client.KafkaEnvConfig{
 			BootstrapServers: obj.Spec.BootstrapServers,
-			Net: source.AdapterNet{
-				SASL: source.AdapterSASL{
+			Net: client.AdapterNet{
+				SASL: client.AdapterSASL{
 					Enable:   obj.Spec.Net.SASL.Enable,
 					User:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.User.SecretKeyRef),
 					Password: a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Password.SecretKeyRef),
 					Type:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Type.SecretKeyRef),
 				},
-				TLS: source.AdapterTLS{
+				TLS: client.AdapterTLS{
 					Enable: obj.Spec.Net.TLS.Enable,
 					Cert:   a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Cert.SecretKeyRef),
 					Key:    a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Key.SecretKeyRef),
diff --git a/pkg/source/reconciler/source/config_watcher.go b/pkg/source/reconciler/source/config_watcher.go
new file mode 100644
index 000000000..50bf39789
--- /dev/null
+++ b/pkg/source/reconciler/source/config_watcher.go
@@ -0,0 +1,169 @@
+/*
+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 source
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/eventing/pkg/reconciler/source"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/logging"
+)
+
+const (
+	EnvKafkaCfg           = "K_KAFKA_CONFIG"
+	kafkaConfigMapNameEnv = "CONFIG_KAFKA_NAME"
+)
+
+type KafkaConfig struct {
+	SaramaYamlString string
+}
+
+type KafkaSourceConfigAccessor interface {
+	source.ConfigAccessor
+	KafkaConfig() *KafkaConfig
+}
+
+// ToEnvVars serializes the contents of the ConfigWatcher to individual
+// environment variables.
+func (cw *KafkaSourceConfigWatcher) ToEnvVars() []corev1.EnvVar {
+	envs := cw.ConfigWatcher.ToEnvVars()
+
+	if cw.KafkaConfig() != nil {
+		envs = append(envs, cw.kafkaConfigEnvVar())
+	}
+
+	return envs
+}
+
+// KafkaConfig returns the logging configuration from the ConfigWatcher.
+func (cw *KafkaSourceConfigWatcher) KafkaConfig() *KafkaConfig {
+	if cw == nil {
+		return nil
+	}
+	return cw.kafkaCfg
+}
+
+var _ KafkaSourceConfigAccessor = (*KafkaSourceConfigWatcher)(nil)
+
+type KafkaSourceConfigWatcher struct {
+	*source.ConfigWatcher
+	logger *zap.SugaredLogger
+
+	kafkaCfg *KafkaConfig
+}
+
+// WatchConfigurations returns a ConfigWatcher initialized with the given
+// options. If no option is passed, the ConfigWatcher observes ConfigMaps for
+// logging, metrics, tracing and Kafka.
+func WatchConfigurations(loggingCtx context.Context, component string,
+	cmw configmap.Watcher) *KafkaSourceConfigWatcher {
+
+	configWatcher := source.WatchConfigurations(loggingCtx, component, cmw)
+
+	cw := &KafkaSourceConfigWatcher{
+		ConfigWatcher: configWatcher,
+		logger:        logging.FromContext(loggingCtx),
+		kafkaCfg:      nil,
+	}
+
+	WatchConfigMapWithKafka(cw, cmw)
+
+	return cw
+}
+
+// WatchConfigMapWithKafka observes a Kafka ConfigMap.
+func WatchConfigMapWithKafka(cw *KafkaSourceConfigWatcher, cmw configmap.Watcher) {
+	cw.kafkaCfg = &KafkaConfig{}
+	watchConfigMap(cmw, KafkaConfigMapName(), cw.updateFromKafkaConfigMap)
+}
+
+func watchConfigMap(cmw configmap.Watcher, cmName string, obs configmap.Observer) {
+	if dcmw, ok := cmw.(configmap.DefaultingWatcher); ok {
+		dcmw.WatchWithDefault(corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{Name: cmName},
+			Data:       map[string]string{},
+		}, obs)
+
+	} else {
+		cmw.Watch(cmName, obs)
+	}
+}
+
+func (cw *KafkaSourceConfigWatcher) updateFromKafkaConfigMap(cfg *corev1.ConfigMap) {
+	kafkaCfg, err := NewKafkaConfigFromConfigMap(cfg)
+	if err != nil {
+		cw.logger.Warnw("ignoring configuration in Kafka configmap ", zap.String("cfg.Name", cfg.Name), zap.Error(err))
+		return
+	}
+
+	cw.kafkaCfg = kafkaCfg
+
+	cw.logger.Debugw("Updated Kafka config from ConfigMap", zap.Any("ConfigMap", cfg))
+	cw.logger.Debugf("Updated Kafka config from ConfigMap 2 %+v", cw.kafkaCfg)
+}
+
+func NewKafkaConfigFromConfigMap(cfg *corev1.ConfigMap) (*KafkaConfig, error) {
+	if cfg == nil {
+		return nil, fmt.Errorf("kafka configmap does not exist")
+	}
+	if _, ok := cfg.Data[constants.SaramaSettingsConfigKey]; !ok {
+		return nil, fmt.Errorf("'%s' key does not exist in Kafka configmap", constants.SaramaSettingsConfigKey)
+	}
+	delete(cfg.Data, "_example")
+	return &KafkaConfig{
+		SaramaYamlString: cfg.Data[constants.SaramaSettingsConfigKey],
+	}, nil
+}
+
+// kafkaConfigEnvVar returns an EnvVar containing the serialized Kafka
+// configuration from the ConfigWatcher.
+func (cw *KafkaSourceConfigWatcher) kafkaConfigEnvVar() corev1.EnvVar {
+	cfg, err := kafkaConfigToJSON(cw.KafkaConfig())
+	if err != nil {
+		cw.logger.Warnw("Error while serializing Kafka config", zap.Error(err))
+	}
+
+	return corev1.EnvVar{
+		Name:  EnvKafkaCfg,
+		Value: cfg,
+	}
+}
+
+func kafkaConfigToJSON(cfg *KafkaConfig) (string, error) { //nolint // for backcompat.
+	if cfg == nil {
+		return "", nil
+	}
+
+	jsonOpts, err := json.Marshal(cfg)
+	return string(jsonOpts), err
+}
+
+// ConfigMapName gets the name of the Kafka ConfigMap
+func KafkaConfigMapName() string {
+	if cm := os.Getenv(kafkaConfigMapNameEnv); cm != "" {
+		return cm
+	}
+	return constants.SettingsConfigMapName
+}
diff --git a/pkg/source/reconciler/source/controller.go b/pkg/source/reconciler/source/controller.go
index 9446d5075..659aee235 100644
--- a/pkg/source/reconciler/source/controller.go
+++ b/pkg/source/reconciler/source/controller.go
@@ -23,8 +23,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	"knative.dev/eventing/pkg/apis/sources/v1alpha1"
-	"knative.dev/eventing/pkg/reconciler/source"
-
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	deploymentinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
 	"knative.dev/pkg/configmap"
@@ -58,7 +56,7 @@ func NewController(
 		deploymentLister:    deploymentInformer.Lister(),
 		receiveAdapterImage: raImage,
 		loggingContext:      ctx,
-		configs:             source.WatchConfigurations(ctx, component, cmw),
+		configs:             WatchConfigurations(ctx, component, cmw),
 	}
 
 	impl := kafkasource.NewImpl(ctx, c)
diff --git a/pkg/source/reconciler/source/kafkasource.go b/pkg/source/reconciler/source/kafkasource.go
index 90338d6b3..2585caafc 100644
--- a/pkg/source/reconciler/source/kafkasource.go
+++ b/pkg/source/reconciler/source/kafkasource.go
@@ -31,7 +31,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing/pkg/reconciler/source"
 	"knative.dev/eventing/pkg/utils"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
@@ -98,7 +97,7 @@ type Reconciler struct {
 
 	sinkResolver *resolver.URIResolver
 
-	configs source.ConfigAccessor
+	configs KafkaSourceConfigAccessor
 }
 
 // Check that our Reconciler implements Interface
diff --git a/vendor/knative.dev/pkg/controller/stats_reporter.go b/vendor/knative.dev/pkg/controller/stats_reporter.go
index 92a5848a1..474bab3fc 100644
--- a/vendor/knative.dev/pkg/controller/stats_reporter.go
+++ b/vendor/knative.dev/pkg/controller/stats_reporter.go
@@ -26,7 +26,6 @@ import (
 	"go.opencensus.io/tag"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/tools/cache"
 	kubemetrics "k8s.io/client-go/tools/metrics"
 	"k8s.io/client-go/util/workqueue"
 	"knative.dev/pkg/metrics"
@@ -34,8 +33,8 @@ import (
 )
 
 var (
-	workQueueDepthStat   = stats.Int64("work_queue_depth", "Depth of the work queue", stats.UnitNone)
-	reconcileCountStat   = stats.Int64("reconcile_count", "Number of reconcile operations", stats.UnitNone)
+	workQueueDepthStat   = stats.Int64("work_queue_depth", "Depth of the work queue", stats.UnitDimensionless)
+	reconcileCountStat   = stats.Int64("reconcile_count", "Number of reconcile operations", stats.UnitDimensionless)
 	reconcileLatencyStat = stats.Int64("reconcile_latency", "Latency of reconcile operations", stats.UnitMilliseconds)
 
 	// reconcileDistribution defines the bucket boundaries for the histogram of reconcile latency metric.
@@ -58,104 +57,51 @@ func init() {
 		Adds: stats.Int64(
 			"workqueue_adds_total",
 			"Total number of adds handled by workqueue",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Depth: stats.Int64(
 			"workqueue_depth",
 			"Current depth of workqueue",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Latency: stats.Float64(
 			"workqueue_queue_latency_seconds",
 			"How long in seconds an item stays in workqueue before being requested.",
-			"s",
+			stats.UnitSeconds,
 		),
 		Retries: stats.Int64(
 			"workqueue_retries_total",
 			"Total number of retries handled by workqueue",
-			"s",
+			stats.UnitDimensionless,
 		),
 		WorkDuration: stats.Float64(
 			"workqueue_work_duration_seconds",
 			"How long in seconds processing an item from workqueue takes.",
-			"s",
+			stats.UnitSeconds,
 		),
 		UnfinishedWorkSeconds: stats.Float64(
 			"workqueue_unfinished_work_seconds",
 			"How long in seconds the outstanding workqueue items have been in flight (total).",
-			"s",
+			stats.UnitSeconds,
 		),
 		LongestRunningProcessorSeconds: stats.Float64(
 			"workqueue_longest_running_processor_seconds",
 			"How long in seconds the longest outstanding workqueue item has been in flight.",
-			"s",
+			stats.UnitSeconds,
 		),
 	}
 	workqueue.SetProvider(wp)
 
-	// Register to receive metrics from kubernetes reflectors (what powers informers)
-	// NOTE: today these don't actually seem to wire up to anything in Kubernetes.
-	rp := &metrics.ReflectorProvider{
-		ItemsInList: stats.Float64(
-			"reflector_items_in_list",
-			"How many items an API list returns to the reflectors",
-			stats.UnitNone,
-		),
-		// TODO(mattmoor): This is not in the latest version, so it will
-		// be removed in a future version.
-		ItemsInMatch: stats.Float64(
-			"reflector_items_in_match",
-			"",
-			stats.UnitNone,
-		),
-		ItemsInWatch: stats.Float64(
-			"reflector_items_in_watch",
-			"How many items an API watch returns to the reflectors",
-			stats.UnitNone,
-		),
-		LastResourceVersion: stats.Float64(
-			"reflector_last_resource_version",
-			"Last resource version seen for the reflectors",
-			stats.UnitNone,
-		),
-		ListDuration: stats.Float64(
-			"reflector_list_duration_seconds",
-			"How long an API list takes to return and decode for the reflectors",
-			stats.UnitNone,
-		),
-		Lists: stats.Int64(
-			"reflector_lists_total",
-			"Total number of API lists done by the reflectors",
-			stats.UnitNone,
-		),
-		ShortWatches: stats.Int64(
-			"reflector_short_watches_total",
-			"Total number of short API watches done by the reflectors",
-			stats.UnitNone,
-		),
-		WatchDuration: stats.Float64(
-			"reflector_watch_duration_seconds",
-			"How long an API watch takes to return and decode for the reflectors",
-			stats.UnitNone,
-		),
-		Watches: stats.Int64(
-			"reflector_watches_total",
-			"Total number of API watches done by the reflectors",
-			stats.UnitNone,
-		),
-	}
-	cache.SetReflectorMetricsProvider(rp)
-
 	cp := &metrics.ClientProvider{
 		Latency: stats.Float64(
 			"client_latency",
 			"How long Kubernetes API requests take",
-			"s",
+			stats.UnitSeconds,
 		),
 		Result: stats.Int64(
 			"client_results",
 			"Total number of API requests (broken down by status code)",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 	}
 	opts := kubemetrics.RegisterOpts{
@@ -181,7 +127,6 @@ func init() {
 		TagKeys:     []tag.Key{reconcilerTagKey, successTagKey, NamespaceTagKey},
 	}}
 	views = append(views, wp.DefaultViews()...)
-	views = append(views, rp.DefaultViews()...)
 	views = append(views, cp.DefaultViews()...)
 
 	// Create views to see our measurements. This can return an error if
diff --git a/vendor/knative.dev/pkg/metrics/memstats.go b/vendor/knative.dev/pkg/metrics/memstats.go
index 719c7e018..d96a0f4bc 100644
--- a/vendor/knative.dev/pkg/metrics/memstats.go
+++ b/vendor/knative.dev/pkg/metrics/memstats.go
@@ -33,112 +33,112 @@ func NewMemStatsAll() *MemStatsProvider {
 		Alloc: stats.Int64(
 			"go_alloc",
 			"The number of bytes of allocated heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		TotalAlloc: stats.Int64(
 			"go_total_alloc",
 			"The cumulative bytes allocated for heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Sys: stats.Int64(
 			"go_sys",
 			"The total bytes of memory obtained from the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Lookups: stats.Int64(
 			"go_lookups",
 			"The number of pointer lookups performed by the runtime.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Mallocs: stats.Int64(
 			"go_mallocs",
 			"The cumulative count of heap objects allocated.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Frees: stats.Int64(
 			"go_frees",
 			"The cumulative count of heap objects freed.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapAlloc: stats.Int64(
 			"go_heap_alloc",
 			"The number of bytes of allocated heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapSys: stats.Int64(
 			"go_heap_sys",
 			"The number of bytes of heap memory obtained from the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapIdle: stats.Int64(
 			"go_heap_idle",
 			"The number of bytes in idle (unused) spans.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapInuse: stats.Int64(
 			"go_heap_in_use",
 			"The number of bytes in in-use spans.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapReleased: stats.Int64(
 			"go_heap_released",
 			"The number of bytes of physical memory returned to the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapObjects: stats.Int64(
 			"go_heap_objects",
 			"The number of allocated heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		StackInuse: stats.Int64(
 			"go_stack_in_use",
 			"The number of bytes in stack spans.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		StackSys: stats.Int64(
 			"go_stack_sys",
 			"The number of bytes of stack memory obtained from the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MSpanInuse: stats.Int64(
 			"go_mspan_in_use",
 			"The number of bytes of allocated mspan structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MSpanSys: stats.Int64(
 			"go_mspan_sys",
 			"The number of bytes of memory obtained from the OS for mspan structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MCacheInuse: stats.Int64(
 			"go_mcache_in_use",
 			"The number of bytes of allocated mcache structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MCacheSys: stats.Int64(
 			"go_mcache_sys",
 			"The number of bytes of memory obtained from the OS for mcache structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		BuckHashSys: stats.Int64(
 			"go_bucket_hash_sys",
 			"The number of bytes of memory in profiling bucket hash tables.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		GCSys: stats.Int64(
 			"go_gc_sys",
 			"The number of bytes of memory in garbage collection metadata.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		OtherSys: stats.Int64(
 			"go_other_sys",
 			"The number of bytes of memory in miscellaneous off-heap runtime allocations.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		NextGC: stats.Int64(
 			"go_next_gc",
 			"The target heap size of the next GC cycle.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		LastGC: stats.Int64(
 			"go_last_gc",
@@ -153,17 +153,17 @@ func NewMemStatsAll() *MemStatsProvider {
 		NumGC: stats.Int64(
 			"go_num_gc",
 			"The number of completed GC cycles.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		NumForcedGC: stats.Int64(
 			"go_num_forced_gc",
 			"The number of GC cycles that were forced by the application calling the GC function.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		GCCPUFraction: stats.Float64(
 			"go_gc_cpu_fraction",
 			"The fraction of this program's available CPU time used by the GC since the program started.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 	}
 }
diff --git a/vendor/knative.dev/pkg/metrics/reflector.go b/vendor/knative.dev/pkg/metrics/reflector.go
deleted file mode 100644
index 548b811b6..000000000
--- a/vendor/knative.dev/pkg/metrics/reflector.go
+++ /dev/null
@@ -1,176 +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 metrics
-
-import (
-	"go.opencensus.io/stats"
-	"go.opencensus.io/stats/view"
-	"go.opencensus.io/tag"
-	"k8s.io/client-go/tools/cache"
-)
-
-// ReflectorProvider implements reflector.MetricsProvider and may be used with
-// reflector.SetProvider to have metrics exported to the provided metrics.
-type ReflectorProvider struct {
-	ItemsInList *stats.Float64Measure
-	// TODO(mattmoor): This is not in the latest version, so it will
-	// be removed in a future version.
-	ItemsInMatch        *stats.Float64Measure
-	ItemsInWatch        *stats.Float64Measure
-	LastResourceVersion *stats.Float64Measure
-	ListDuration        *stats.Float64Measure
-	Lists               *stats.Int64Measure
-	ShortWatches        *stats.Int64Measure
-	WatchDuration       *stats.Float64Measure
-	Watches             *stats.Int64Measure
-}
-
-var (
-	_ cache.MetricsProvider = (*ReflectorProvider)(nil)
-)
-
-// NewItemsInListMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewItemsInListMetric(name string) cache.SummaryMetric {
-	return floatMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.ItemsInList,
-	}
-}
-
-// ItemsInListView returns a view of the ItemsInList metric.
-func (rp *ReflectorProvider) ItemsInListView() *view.View {
-	return measureView(rp.ItemsInList, view.Distribution(BucketsNBy10(0.1, 6)...))
-}
-
-// NewItemsInMatchMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewItemsInMatchMetric(name string) cache.SummaryMetric {
-	return floatMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.ItemsInMatch,
-	}
-}
-
-// ItemsInMatchView returns a view of the ItemsInMatch metric.
-func (rp *ReflectorProvider) ItemsInMatchView() *view.View {
-	return measureView(rp.ItemsInMatch, view.Distribution(BucketsNBy10(0.1, 6)...))
-}
-
-// NewItemsInWatchMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewItemsInWatchMetric(name string) cache.SummaryMetric {
-	return floatMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.ItemsInWatch,
-	}
-}
-
-// ItemsInWatchView returns a view of the ItemsInWatch metric.
-func (rp *ReflectorProvider) ItemsInWatchView() *view.View {
-	return measureView(rp.ItemsInWatch, view.Distribution(BucketsNBy10(0.1, 6)...))
-}
-
-// NewLastResourceVersionMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewLastResourceVersionMetric(name string) cache.GaugeMetric {
-	return floatMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.LastResourceVersion,
-	}
-}
-
-// LastResourceVersionView returns a view of the LastResourceVersion metric.
-func (rp *ReflectorProvider) LastResourceVersionView() *view.View {
-	return measureView(rp.LastResourceVersion, view.LastValue())
-}
-
-// NewListDurationMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewListDurationMetric(name string) cache.SummaryMetric {
-	return floatMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.ListDuration,
-	}
-}
-
-// ListDurationView returns a view of the ListDuration metric.
-func (rp *ReflectorProvider) ListDurationView() *view.View {
-	return measureView(rp.ListDuration, view.Distribution(BucketsNBy10(0.1, 6)...))
-}
-
-// NewListsMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewListsMetric(name string) cache.CounterMetric {
-	return counterMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.Lists,
-	}
-}
-
-// ListsView returns a view of the Lists metric.
-func (rp *ReflectorProvider) ListsView() *view.View {
-	return measureView(rp.Lists, view.Count())
-}
-
-// NewShortWatchesMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewShortWatchesMetric(name string) cache.CounterMetric {
-	return counterMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.ShortWatches,
-	}
-}
-
-// ShortWatchesView returns a view of the ShortWatches metric.
-func (rp *ReflectorProvider) ShortWatchesView() *view.View {
-	return measureView(rp.ShortWatches, view.Count())
-}
-
-// NewWatchDurationMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewWatchDurationMetric(name string) cache.SummaryMetric {
-	return floatMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.WatchDuration,
-	}
-}
-
-// WatchDurationView returns a view of the WatchDuration metric.
-func (rp *ReflectorProvider) WatchDurationView() *view.View {
-	return measureView(rp.WatchDuration, view.Distribution(BucketsNBy10(0.1, 6)...))
-}
-
-// NewWatchesMetric implements MetricsProvider
-func (rp *ReflectorProvider) NewWatchesMetric(name string) cache.CounterMetric {
-	return counterMetric{
-		mutators: []tag.Mutator{tag.Insert(tagName, name)},
-		measure:  rp.Watches,
-	}
-}
-
-// WatchesView returns a view of the Watches metric.
-func (rp *ReflectorProvider) WatchesView() *view.View {
-	return measureView(rp.Watches, view.Count())
-}
-
-// DefaultViews returns a list of views suitable for passing to view.Register
-func (rp *ReflectorProvider) DefaultViews() []*view.View {
-	return []*view.View{
-		rp.ItemsInListView(),
-		rp.ItemsInMatchView(),
-		rp.ItemsInWatchView(),
-		rp.LastResourceVersionView(),
-		rp.ListDurationView(),
-		rp.ListsView(),
-		rp.ShortWatchesView(),
-		rp.WatchDurationView(),
-		rp.WatchesView(),
-	}
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 4e9c6ef97..35d90e72e 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1080,7 +1080,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.20.1-0.20210211204829-1c771ce4058e
+# knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1218,7 +1218,7 @@ knative.dev/eventing/test/test_images/recordevents
 # knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe
+# knative.dev/pkg v0.0.0-20210216013737-584933f8280b
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/pkg/apis/sources/v1beta1/kafka_validation_test.go b/pkg/apis/sources/v1beta1/kafka_validation_test.go
index 1c47cd5d7..aac512c87 100644
--- a/pkg/apis/sources/v1beta1/kafka_validation_test.go
+++ b/pkg/apis/sources/v1beta1/kafka_validation_test.go
@@ -29,6 +29,14 @@ var (
 	fullSpec = KafkaSourceSpec{
 		KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
 			BootstrapServers: []string{"servers"},
+			Net: bindingsv1beta1.KafkaNetSpec{
+				SASL: bindingsv1beta1.KafkaSASLSpec{
+					Enable: false,
+				},
+				TLS: bindingsv1beta1.KafkaTLSSpec{
+					Enable: false,
+				},
+			},
 		},
 		Topics:        []string{"topics"},
 		ConsumerGroup: "group",
@@ -44,6 +52,64 @@ var (
 	}
 )
 
+func TestKafkaSourceCheckRequiredFields(t *testing.T) {
+	testCases := map[string]struct {
+		orig    *KafkaSourceSpec
+		allowed bool
+	}{
+		"nil original": {
+			orig:    &KafkaSourceSpec{},
+			allowed: false,
+		},
+		"nil topic": {
+			allowed: false,
+			orig: &KafkaSourceSpec{
+				KafkaAuthSpec: fullSpec.KafkaAuthSpec,
+				SourceSpec:    fullSpec.SourceSpec,
+			},
+		},
+		"nil bootstrapServer": {
+			orig: &KafkaSourceSpec{
+				Topics:     fullSpec.Topics,
+				SourceSpec: fullSpec.SourceSpec,
+			},
+			allowed: false,
+		},
+		"nil sink": {
+			orig: &KafkaSourceSpec{
+				KafkaAuthSpec: fullSpec.KafkaAuthSpec,
+				Topics:        fullSpec.Topics,
+			},
+			allowed: false,
+		},
+		"min required fields": {
+			orig: &KafkaSourceSpec{
+				KafkaAuthSpec: fullSpec.KafkaAuthSpec,
+				Topics:        fullSpec.Topics,
+				SourceSpec:    fullSpec.SourceSpec,
+			},
+			allowed: true,
+		},
+		"full source": {
+			orig:    &fullSpec,
+			allowed: true,
+		},
+	}
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			ctx := context.TODO()
+			orig := &KafkaSource{
+				Spec: *tc.orig,
+			}
+			ctx = apis.WithinCreate(ctx)
+			err := orig.Validate(ctx)
+			if tc.allowed != (err == nil) {
+				t.Fatalf("Required Field Not Included: %v", err)
+			}
+		})
+	}
+}
+
 func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 	testCases := map[string]struct {
 		orig    *KafkaSourceSpec
@@ -77,7 +143,8 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 		"Sink.APIVersion changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -89,12 +156,13 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"Sink.Kind changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -106,12 +174,13 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"Sink.Namespace changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -123,12 +192,13 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"Sink.Name changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -140,7 +210,7 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"ServiceAccountName changed": {
 			orig: &fullSpec,
@@ -164,6 +234,45 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 			updated: fullSpec,
 			allowed: true,
 		},
+		"consumerGroup changed": {
+			orig: &fullSpec,
+			updated: KafkaSourceSpec{
+				ConsumerGroup: "no-way",
+			},
+			allowed: false,
+		},
+		"Kafka TLS Spec changed": {
+			orig: &fullSpec,
+			updated: KafkaSourceSpec{
+				ConsumerGroup: fullSpec.ConsumerGroup,
+				SourceSpec:    fullSpec.SourceSpec,
+				Topics:        fullSpec.Topics,
+				KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+					Net: bindingsv1beta1.KafkaNetSpec{
+						TLS: bindingsv1beta1.KafkaTLSSpec{
+							Enable: true,
+						},
+					},
+				},
+			},
+			allowed: true,
+		},
+		"Kafka SASL Spec changed": {
+			orig: &fullSpec,
+			updated: KafkaSourceSpec{
+				ConsumerGroup: fullSpec.ConsumerGroup,
+				SourceSpec:    fullSpec.SourceSpec,
+				Topics:        fullSpec.Topics,
+				KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+					Net: bindingsv1beta1.KafkaNetSpec{
+						SASL: bindingsv1beta1.KafkaSASLSpec{
+							Enable: true,
+						},
+					},
+				},
+			},
+			allowed: true,
+		},
 	}
 
 	for n, tc := range testCases {
diff --git a/pkg/source/client_test.go b/pkg/source/client/client_test.go
similarity index 96%
rename from pkg/source/client_test.go
rename to pkg/source/client/client_test.go
index cee49af9f..0d8fa4a10 100644
--- a/pkg/source/client_test.go
+++ b/pkg/source/client/client_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package source
+package client
 
 import (
 	"context"
@@ -22,7 +22,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
-	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
+	"knative.dev/eventing-kafka/pkg/common/client"
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
 
@@ -155,7 +155,7 @@ func TestNewConfig(t *testing.T) {
 			for k, v := range tc.env {
 				_ = os.Setenv(k, v)
 			}
-			servers, config, err := NewConfig(context.Background())
+			servers, config, err := NewConfigFromEnv(context.Background())
 			if err != nil && tc.wantErr != true {
 				t.Fatal(err)
 			}
@@ -200,7 +200,7 @@ func TestAdminClient(t *testing.T) {
 	})
 
 	// mock broker does not support TLS ...
-	admin, err := MakeAdminClient(ctx, "test-client", nil, &utils.KafkaConfig{Brokers: []string{seedBroker.Addr()}})
+	admin, err := client.MakeAdminClient(ctx, "test-client", nil, "", []string{seedBroker.Addr()})
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/pkg/source/reconciler/source/config_watcher_test.go b/pkg/source/reconciler/source/config_watcher_test.go
new file mode 100644
index 000000000..f852d1d5b
--- /dev/null
+++ b/pkg/source/reconciler/source/config_watcher_test.go
@@ -0,0 +1,193 @@
+/*
+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 source
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing/pkg/reconciler/source"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/metrics"
+	tracingconfig "knative.dev/pkg/tracing/config"
+
+	loggingtesting "knative.dev/pkg/logging/testing"
+
+	// metrics package is suggesting to the following
+	// import to fix the panics in unit tests
+	_ "knative.dev/pkg/metrics/testing"
+)
+
+const testComponent = "test_component"
+
+func TestNewConfigWatcher_defaults(t *testing.T) {
+	testCases := []struct {
+		name                  string
+		cmw                   configmap.Watcher
+		expectLoggingContains string
+		expectMetricsContains string
+		expectTracingContains string
+		expectKafkaContains   string
+	}{
+		{
+			name:                  "With pre-filled sample data",
+			cmw:                   configMapWatcherWithSampleData(),
+			expectLoggingContains: `"zap-logger-config":"{\"level\": \"fatal\"}"`,
+			expectMetricsContains: `"ConfigMap":{"metrics.backend":"test"}`,
+			expectTracingContains: `"zipkin-endpoint":"zipkin.test"`,
+			expectKafkaContains:   `"SaramaYamlString":"{Version: 2.0.0}"`,
+		},
+		{
+			name: "With empty data",
+			cmw:  configMapWatcherWithEmptyData(),
+			// logging defaults to Knative's defaults
+			expectLoggingContains: ``,
+			// metrics defaults to empty ConfigMap
+			expectMetricsContains: `"ConfigMap":{}`,
+			// tracing defaults to None backend
+			expectTracingContains: `"backend":"none"`,
+			// kafka defaults to empty
+			expectKafkaContains: ``,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := loggingtesting.TestContextWithLogger(t)
+			cw := WatchConfigurations(ctx, testComponent, tc.cmw)
+
+			assert.NotNil(t, cw.LoggingConfig(), "logging config should be enabled")
+			assert.NotNil(t, cw.MetricsConfig(), "metrics config should be enabled")
+			assert.NotNil(t, cw.TracingConfig(), "tracing config should be enabled")
+			assert.NotNil(t, cw.KafkaConfig(), "kafka config should be enabled")
+
+			envs := cw.ToEnvVars()
+
+			const expectEnvs = 4
+			require.Lenf(t, envs, expectEnvs, "there should be %d env var(s)", expectEnvs)
+
+			assert.Equal(t, source.EnvLoggingCfg, envs[0].Name, "first env var is logging config")
+			assert.Contains(t, envs[0].Value, tc.expectLoggingContains)
+
+			assert.Equal(t, source.EnvMetricsCfg, envs[1].Name, "second env var is metrics config")
+			assert.Contains(t, envs[1].Value, tc.expectMetricsContains)
+
+			assert.Equal(t, source.EnvTracingCfg, envs[2].Name, "third env var is tracing config")
+			assert.Contains(t, envs[2].Value, tc.expectTracingContains)
+
+			assert.Equal(t, EnvKafkaCfg, envs[3].Name, "third env var is kafka config")
+			assert.Contains(t, envs[3].Value, tc.expectKafkaContains)
+		})
+	}
+}
+
+// configMapWatcherWithSampleData constructs a Watcher for static sample data.
+func configMapWatcherWithSampleData() configmap.Watcher {
+	return configmap.NewStaticWatcher(
+		newTestConfigMap(logging.ConfigMapName(), loggingConfigMapData()),
+		newTestConfigMap(metrics.ConfigMapName(), metricsConfigMapData()),
+		newTestConfigMap(tracingconfig.ConfigName, tracingConfigMapData()),
+		newTestConfigMap(KafkaConfigMapName(), kafkaConfigMapData()),
+	)
+}
+
+// configMapWatcherWithEmptyData constructs a Watcher for empty data.
+func configMapWatcherWithEmptyData() configmap.Watcher {
+	return configmap.NewStaticWatcher(
+		newTestConfigMap(logging.ConfigMapName(), nil),
+		newTestConfigMap(metrics.ConfigMapName(), nil),
+		newTestConfigMap(tracingconfig.ConfigName, nil),
+		newTestConfigMap(KafkaConfigMapName(), nil),
+	)
+}
+
+func newTestConfigMap(name string, data map[string]string) *corev1.ConfigMap {
+	if data == nil {
+		data = make(map[string]string, 1)
+	}
+
+	// _example key is always appended to mimic Knative's release manifests
+	data["_example"] = "test-config"
+
+	return &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Data: data,
+	}
+}
+
+// ConfigMap data generators
+func loggingConfigMapData() map[string]string {
+	return map[string]string{
+		"zap-logger-config": `{"level": "fatal"}`,
+	}
+}
+func metricsConfigMapData() map[string]string {
+	return map[string]string{
+		"metrics.backend": "test",
+	}
+}
+func tracingConfigMapData() map[string]string {
+	return map[string]string{
+		"backend":         "zipkin",
+		"zipkin-endpoint": "zipkin.test",
+	}
+}
+
+func kafkaConfigMapData() map[string]string {
+	return map[string]string{
+		"sarama": `{Version: 2.0.0}`,
+	}
+}
+
+func TestKafkaConfigToJSON(t *testing.T) {
+
+	testCases := []struct {
+		name     string
+		cfg      KafkaConfig
+		success  bool
+		expected string
+	}{{
+		name: "simple marshal",
+		cfg: KafkaConfig{
+			SaramaYamlString: `Version: 2.0.0`,
+		},
+		success:  true,
+		expected: `{"SaramaYamlString":"Version: 2.0.0"}`,
+	}}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			output, err := kafkaConfigToJSON(&tc.cfg)
+			if tc.success == true && err != nil {
+				t.Errorf("KafkaConfig should've been converted to JSON successfully. Err: %e", err)
+			}
+			if tc.success == false && err != nil {
+				t.Errorf("KafkaConfig should not have been converted to JSON successfully.")
+			}
+			if output != tc.expected {
+				t.Errorf("KafkaConfig conversion is wrong, want: %s vs got %s", tc.expected, output)
+			}
+		})
+	}
+
+}
diff --git a/test/README.md b/test/README.md
index e2bc37b04..74c148a68 100644
--- a/test/README.md
+++ b/test/README.md
@@ -57,6 +57,8 @@ If you have set up a running environment that meets
 [the e2e test environment requirements](#environment-requirements), you can run
 with `test/e2e-tests.sh --run-tests --skip-knative-setup`.
 
+If you want to skip tearing down your cluster, add the `--skip-teardowns` flag.
+
 ## Running tests with `go test` command
 
 ### Running unit tests
@@ -98,6 +100,13 @@ bevavior you can specify a different tag through `-tag`:
 SYSTEM_NAMESPACE=knative-eventing go test -v -tags=e2e,source -count=1 ./test/e2e -tag e2e
 ```
 
+To run the e2e source tests against the multi-tenant source adapter you need to
+specify the build tags `source` and `mtsource`:
+
+```bash
+SYSTEM_NAMESPACE=knative-eventing go test -v -tags=source,mtsource  ./test/e2e
+```
+
 #### One test case
 
 To run one e2e test case, e.g. `TestSingleBinaryEventForChannel`, use
diff --git a/test/e2e/kafka_source_test.go b/test/e2e/kafka_source_test.go
index 8f3f385bf..d37128244 100644
--- a/test/e2e/kafka_source_test.go
+++ b/test/e2e/kafka_source_test.go
@@ -422,3 +422,163 @@ func mustJsonMarshal(t *testing.T, val interface{}) string {
 	}
 	return string(data)
 }
+
+type message struct {
+	cloudEventType string
+	payload        []byte
+	headers        map[string]string
+	key            string
+}
+
+type updateTest struct {
+	auth      authSetup
+	topicName string
+	sinkName  string
+}
+
+var (
+	defaultKafkaSource = updateTest{
+		auth: authSetup{
+			bootStrapServer: kafkaBootstrapUrlTLS,
+			SASLEnabled:     false,
+			TLSEnabled:      true,
+		},
+		topicName: "initial-topic",
+		sinkName:  "default-event-recorder",
+	}
+)
+
+func TestKafkaSourceUpdate(t *testing.T) {
+
+	testCases := map[string]updateTest{
+		"no-change": defaultKafkaSource,
+		"change-sink": {
+			auth:      defaultKafkaSource.auth,
+			topicName: defaultKafkaSource.topicName,
+			sinkName:  "update-event-recorder",
+		},
+		"change-topic": {
+			auth:      defaultKafkaSource.auth,
+			topicName: "update-topic",
+			sinkName:  defaultKafkaSource.sinkName,
+		},
+		"change-bootstrap-server": {
+			auth: authSetup{
+				bootStrapServer: kafkaBootstrapUrlPlain,
+				TLSEnabled:      false,
+				SASLEnabled:     false,
+			},
+			topicName: defaultKafkaSource.topicName,
+			sinkName:  defaultKafkaSource.sinkName,
+		},
+	}
+
+	for name, test := range testCases {
+		test := test
+		t.Run(name, func(t *testing.T) {
+			testKafkaSourceUpdate(t, name, test)
+		})
+	}
+}
+
+func testKafkaSourceUpdate(t *testing.T, name string, test updateTest) {
+
+	messagePayload := []byte(`{"value":5}`)
+	matcherGen := func(cloudEventsSourceName, originalOrUpdate string) EventMatcher {
+		return AllOf(
+			HasSource(cloudEventsSourceName),
+			HasData(messagePayload),
+			HasExtension("kafkaheaderceoriginalorupdate", originalOrUpdate))
+	}
+
+	originalMessage := message{
+		payload: messagePayload,
+		headers: map[string]string{
+			"ce_originalorupdate": "original",
+		},
+		key: "0",
+	}
+	updateMessage := message{
+		payload: messagePayload,
+		headers: map[string]string{
+			"ce_originalorupdate": "update",
+		},
+		key: "0",
+	}
+
+	client := testlib.Setup(t, true)
+	defer testlib.TearDown(client)
+
+	t.Logf("Creating topic: %s\n", defaultKafkaSource.topicName+name)
+	helpers.MustCreateTopic(client, kafkaClusterName, kafkaClusterNamespace, defaultKafkaSource.topicName+name)
+
+	t.Logf("Copying secrets: %s\n", defaultKafkaSource.topicName+name)
+	_, err := utils.CopySecret(client.Kube.CoreV1(), "knative-eventing", kafkaTLSSecret, client.Namespace, "default")
+	if err != nil {
+		t.Fatalf("could not copy secret(%s): %v", kafkaTLSSecret, err)
+	}
+
+	t.Logf("Creating default eventrecorder pod: %s\n", defaultKafkaSource.sinkName)
+	originalEventTracker, _ := recordevents.StartEventRecordOrFail(context.Background(), client, defaultKafkaSource.sinkName)
+	client.WaitForAllTestResourcesReadyOrFail(context.Background())
+
+	kafkaSourceName := "e2e-kafka-source-" + name
+
+	t.Logf("Creating kafkasource: %s\n", kafkaSourceName)
+	contribtestlib.CreateKafkaSourceV1Beta1OrFail(client, contribresources.KafkaSourceV1Beta1(
+		defaultKafkaSource.auth.bootStrapServer,
+		defaultKafkaSource.topicName+name,
+		resources.ServiceRef(defaultKafkaSource.sinkName),
+		contribresources.WithNameV1Beta1(kafkaSourceName),
+		withAuthEnablementV1Beta1(defaultKafkaSource.auth),
+	))
+	client.WaitForAllTestResourcesReadyOrFail(context.Background())
+
+	t.Logf("Send update event to kafkatopic")
+	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain,
+		defaultKafkaSource.topicName+name,
+		originalMessage.key, originalMessage.headers, string(originalMessage.payload))
+	eventSourceName := sourcesv1beta1.KafkaEventSource(client.Namespace, kafkaSourceName, defaultKafkaSource.topicName+name)
+	originalEventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(eventSourceName, "original")))
+	t.Logf("Properly received original event for %s\n", eventSourceName)
+
+	var (
+		ksObj               *sourcesv1beta1.KafkaSource
+		newSinkEventTracker *recordevents.EventInfoStore
+	)
+	ksObj = contribtestlib.GetKafkaSourceV1Beta1OrFail(client, kafkaSourceName)
+	if ksObj == nil {
+		t.Fatalf("Unabled to Get kafkasource: %s/%s\n", client.Namespace, kafkaSourceName)
+	}
+	if test.topicName != defaultKafkaSource.topicName {
+		helpers.MustCreateTopic(client, kafkaClusterName, kafkaClusterNamespace, test.topicName+name)
+		ksObj.Spec.Topics = []string{test.topicName + name}
+		eventSourceName = sourcesv1beta1.KafkaEventSource(client.Namespace, kafkaSourceName, test.topicName+name)
+	}
+	if test.sinkName != defaultKafkaSource.sinkName {
+		kSinkRef := resources.ServiceKRef(test.sinkName)
+		ksObj.Spec.Sink.Ref = kSinkRef
+
+		newSinkEventTracker, _ = recordevents.StartEventRecordOrFail(context.Background(), client, test.sinkName)
+	}
+	if test.auth.bootStrapServer != defaultKafkaSource.auth.bootStrapServer {
+		ksObj.Spec.KafkaAuthSpec.BootstrapServers = []string{test.auth.bootStrapServer}
+		ksObj.Spec.KafkaAuthSpec.Net.TLS.Enable = test.auth.TLSEnabled
+		ksObj.Spec.KafkaAuthSpec.Net.SASL.Enable = test.auth.SASLEnabled
+	}
+
+	contribtestlib.UpdateKafkaSourceV1Beta1OrFail(client, ksObj)
+	client.WaitForAllTestResourcesReadyOrFail(context.Background())
+
+	t.Logf("Send update event to kafkatopic")
+	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain,
+		test.topicName+name,
+		updateMessage.key, updateMessage.headers, string(updateMessage.payload))
+
+	if test.sinkName != defaultKafkaSource.sinkName {
+		newSinkEventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(eventSourceName, "update")))
+	} else {
+		originalEventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(eventSourceName, "update")))
+	}
+
+}
diff --git a/test/lib/creation.go b/test/lib/creation.go
index bddcf311b..01e281d2f 100644
--- a/test/lib/creation.go
+++ b/test/lib/creation.go
@@ -58,6 +58,33 @@ func CreateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1bet
 	}
 }
 
+func GetKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource string) *sourcesv1beta1.KafkaSource {
+	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
+	}
+
+	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
+	if ksObj, err := kSources.Get(context.Background(), kafkaSource, metav1.GetOptions{}); err != nil {
+		c.T.Fatalf("Failed to get v1beta1 KafkaSource %q: %v", kafkaSource, err)
+	} else {
+		return ksObj
+	}
+	return nil
+}
+
+func UpdateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1beta1.KafkaSource) {
+	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
+	}
+
+	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
+	if _, err := kSources.Update(context.Background(), kafkaSource, metav1.UpdateOptions{}); err != nil {
+		c.T.Fatalf("Failed to update v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
+	}
+}
+
 func CreateKafkaBindingV1Alpha1OrFail(c *testlib.Client, kafkaBinding *bindingsv1alpha1.KafkaBinding) {
 	kafkaBindingClientSet, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
diff --git a/test/test_images/kafka-publisher/main.go b/test/test_images/kafka-publisher/main.go
index 3de776408..bfdeb2a32 100644
--- a/test/test_images/kafka-publisher/main.go
+++ b/test/test_images/kafka-publisher/main.go
@@ -19,7 +19,7 @@ package main
 import (
 	"log"
 
-	"knative.dev/eventing-kafka/pkg/source"
+	kafkasourceclient "knative.dev/eventing-kafka/pkg/source/client"
 
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
@@ -45,7 +45,7 @@ func main() {
 	log.Printf("GOT: %v", s)
 
 	// Create a Kafka client from our Binding.
-	client, err := source.NewProducer(ctx)
+	client, err := kafkasourceclient.NewProducer(ctx)
 	if err != nil {
 		log.Fatal(err.Error())
 	}