Repository: knative-sandbox/eventing-kafka
Pull request #419 (2021-02-26 18:44:08)

add mtsource owners
/hold

Depends on https://github.com/knative/community/pull/498

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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **lionelvillard** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @lionelvillard` 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":["lionelvillard"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/419?src=pr&el=h1) Report
> Merging [#419](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/419?src=pr&el=desc) (c9a6f8a) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/c7ab439c66df5641d312c59731a8d33ac0bb8b1a?el=desc) (c7ab439) will **decrease** coverage by `0.16%`.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff             @@
##           master     #419      +/-   ##
==========================================
- Coverage   73.13%   72.97%   -0.17%     
==========================================
  Files         129      129              
  Lines        4973     4973              
==========================================
- Hits         3637     3629       -8     
- Misses       1100     1109       +9     
+ Partials      236      235       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/419?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/common/scheduler/statefulset/autoscaler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/419/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvYXV0b3NjYWxlci5nbw==) | `44.44% <0.00%> (-17.78%)` | :arrow_down: |

------

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

lionelvillard: /unhold
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/419#pullrequestreview-599918212" title="Approved">devguyio</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **lionelvillard** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @lionelvillard` 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":["lionelvillard"]} -->
devguyio: LGTM

Patch:
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index bda6c1d2d..0eb8f477f 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -1,9 +1,12 @@
 # Default owners are TOC, release leads and Eventing writers
-* @knative-sandbox/technical-oversight-committee
-* @knative-sandbox/knative-release-leads
-* @knative-sandbox/eventing-writers
-* @knative-sandbox/eventing-kafka-approvers
+* @knative-sandbox/eventing-kafka-approvers @knative-sandbox/eventing-writers @knative-sandbox/knative-release-leads @knative-sandbox/technical-oversight-committee
+
+# mtsource owners
+/cmd/config/source/multi/** @knative-sandbox/eventing-kafka-mtsource-approvers
+/cmd/source/mt* @knative-sandbox/eventing-kafka-mtsource-approvers
+/pkg/common/scheduler/** @knative-sandbox/eventing-kafka-mtsource-approvers
+/pkg/source/** @knative-sandbox/eventing-kafka-mtsource-approvers
 
 # hack and test are owned by productivity
-/hack/* @knative-sandbox/productivity-wg-leads
-/test/* @knative-sandbox/productivity-wg-leads
+/hack/* @knative-sandbox/productivity-wg-leads @knative-sandbox/eventing-kafka-approvers @knative-sandbox/eventing-writers @knative-sandbox/knative-release-leads @knative-sandbox/technical-oversight-committee
+/test/* @knative-sandbox/productivity-wg-leads @knative-sandbox/eventing-kafka-approvers @knative-sandbox/eventing-writers @knative-sandbox/knative-release-leads @knative-sandbox/technical-oversight-committee
diff --git a/go.mod b/go.mod
index 8ecde2d3c..5c53514d4 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.21.0
+	knative.dev/eventing v0.21.1-0.20210302004649-f22066824960
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
-	knative.dev/pkg v0.0.0-20210216013737-584933f8280b
+	knative.dev/pkg v0.0.0-20210226182947-9039dc189ced
 )
diff --git a/go.sum b/go.sum
index 67badb039..f9f7aee0a 100644
--- a/go.sum
+++ b/go.sum
@@ -1244,14 +1244,15 @@ 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.21.0 h1:oZoL0lXXslUClNvoi5mLO2YI6zn2C8qg80mz8Y6Vmkg=
-knative.dev/eventing v0.21.0/go.mod h1:JjbVEOTJJHqo9CTxbTfrMn018hG8fOr3UfBoCJ7KWaA=
+knative.dev/eventing v0.21.1-0.20210302004649-f22066824960 h1:E5lX5F6PNv2GwBcrJyrq+uHls2SnKLX4TLH1BeLq6j4=
+knative.dev/eventing v0.21.1-0.20210302004649-f22066824960/go.mod h1:KtN2FEhPzxyqXKxpJufdq0lzL3Iu+5qjFqBal8Lg2tQ=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf h1:u4cY4jr2LYvhoz/1HBWEPsMiLkm0HMdDTfmmw1RE8zE=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/pkg v0.0.0-20210215165523-84c98f3c3e7a/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/pkg v0.0.0-20210216013737-584933f8280b h1:2v+GBBenjPXhtk3KX/YqxCQVBU6o/AaRkkqE58lgFRQ=
-knative.dev/pkg v0.0.0-20210216013737-584933f8280b/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/reconciler-test v0.0.0-20210216030508-77f50054d024/go.mod h1:RP/K5xJylB72Go6eAsXYEsQHp4zCCNMNjmsqhvq7wko=
+knative.dev/pkg v0.0.0-20210225221047-07b5ddfaea60/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
+knative.dev/pkg v0.0.0-20210226182947-9039dc189ced h1:OCtTbXsJhiR35qCDxFj9Xc8sU3i87ePbc7YlHCt0al4=
+knative.dev/pkg v0.0.0-20210226182947-9039dc189ced/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
+knative.dev/reconciler-test v0.0.0-20210223181346-209f32c8cdc4/go.mod h1:RP/K5xJylB72Go6eAsXYEsQHp4zCCNMNjmsqhvq7wko=
 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/channel/distributed/common/kafka/sarama/sarama.go b/pkg/channel/distributed/common/kafka/sarama/sarama.go
index 08f5ac1c3..3bb7e0cdd 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama.go
@@ -118,3 +118,23 @@ func AuthFromSarama(config *sarama.Config) *client.KafkaAuthConfig {
 		return nil
 	}
 }
+
+// Utility function to convert []byte headers to string ones for logging purposes
+func StringifyHeaders(headers []sarama.RecordHeader) map[string][]string {
+	stringHeaders := make(map[string][]string)
+	for _, header := range headers {
+		key := string(header.Key)
+		stringHeaders[key] = append(stringHeaders[key], string(header.Value))
+	}
+	return stringHeaders
+}
+
+// Pointer-version of the StringifyHeaders function
+func StringifyHeaderPtrs(headers []*sarama.RecordHeader) map[string][]string {
+	stringHeaders := make(map[string][]string)
+	for _, header := range headers {
+		key := string(header.Key)
+		stringHeaders[key] = append(stringHeaders[key], string(header.Value))
+	}
+	return stringHeaders
+}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/handler.go b/pkg/channel/distributed/dispatcher/dispatcher/handler.go
index 13b4da3f1..153701d96 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/handler.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/handler.go
@@ -21,6 +21,8 @@ import (
 	"errors"
 	"net/url"
 
+	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
+
 	"github.com/Shopify/sarama"
 	kafkasaramaprotocol "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -122,9 +124,9 @@ func (h *Handler) consumeMessage(context context.Context, consumerMessage *saram
 
 	// Debug Log Kafka ConsumerMessage
 	if h.Logger.Core().Enabled(zap.DebugLevel) {
-		// Checked Logging Level First To Avoid Calling zap.Any In Production
+		// Checked Logging Level First To Avoid Calling StringifyHeaderPtrs In Production
 		h.Logger.Debug("Consuming Kafka Message",
-			zap.Any("Headers", consumerMessage.Headers),
+			zap.Any("Headers", kafkasarama.StringifyHeaderPtrs(consumerMessage.Headers)), // Log human-readable strings, not base64
 			zap.ByteString("Key", consumerMessage.Key),
 			zap.ByteString("Value", consumerMessage.Value),
 			zap.String("Topic", consumerMessage.Topic),
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index 6b9ffe764..566375869 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -124,8 +124,12 @@ func (p *Producer) ProduceKafkaMessage(ctx context.Context, channelReference eve
 
 	// Produce The Kafka Message To The Kafka Topic
 	if logger.Core().Enabled(zap.DebugLevel) {
-		// Checked Logging Level First To Avoid Calling zap.Any In Production
-		logger.Debug("Producing Kafka Message", zap.Any("Headers", producerMessage.Headers), zap.Any("Message", producerMessage.Value))
+		// Checked Logging Level First To Avoid Calling StringifyHeaders and Encode Functions In Production
+		// There isn't actually any way for the sarama ByteEncoder to return an error and this is debug logging, so ignore the return value
+		msgBytes, _ := producerMessage.Value.Encode()
+		logger.Debug("Producing Kafka Message",
+			zap.Any("Headers", kafkasarama.StringifyHeaders(producerMessage.Headers)), // Log human-readable strings, not base64
+			zap.ByteString("Message", msgBytes))
 	}
 	partition, offset, err := p.kafkaProducer.SendMessage(producerMessage)
 	if err != nil {
diff --git a/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go b/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go
index 7dbd0b894..22e8cc184 100644
--- a/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go
+++ b/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/reconciler.go b/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/reconciler.go
index 9ca9870b3..9dc622014 100644
--- a/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/reconciler.go
+++ b/pkg/client/injection/reconciler/bindings/v1alpha1/kafkabinding/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go b/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go
index ce7675ff9..8e82da2bf 100644
--- a/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go
+++ b/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/reconciler.go b/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/reconciler.go
index 40ccc7af6..d8b567f6c 100644
--- a/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/reconciler.go
+++ b/pkg/client/injection/reconciler/bindings/v1beta1/kafkabinding/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go b/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go
index 712a24d9d..a29a4e238 100644
--- a/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go
+++ b/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/reconciler.go b/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/reconciler.go
index bd435979e..4e288a0a2 100644
--- a/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/reconciler.go
+++ b/pkg/client/injection/reconciler/messaging/v1alpha1/kafkachannel/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go b/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
index 9cffd8d93..c669441af 100644
--- a/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
+++ b/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/reconciler.go b/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/reconciler.go
index fbe3f339a..9e1e06d06 100644
--- a/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/reconciler.go
+++ b/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go b/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go
index 7d9b082de..967880807 100644
--- a/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go
+++ b/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/reconciler.go b/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/reconciler.go
index 087f904d2..c06b06774 100644
--- a/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/reconciler.go
+++ b/pkg/client/injection/reconciler/sources/v1alpha1/kafkasource/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go b/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
index d22f53c31..abcab4bbd 100644
--- a/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
+++ b/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
@@ -113,6 +113,9 @@ func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsF
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go b/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go
index 89758b9cf..5731e8454 100644
--- a/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go
+++ b/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go
@@ -169,6 +169,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
index 1d5ac7203..47332745f 100644
--- a/pkg/common/scheduler/statefulset/autoscaler.go
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -75,14 +75,12 @@ func (s *StatefulSetScheduler) autoscale(ctx context.Context) {
 				// Desired ratio is 0.5 (TODO: configurable)
 				new := int32(math.Ceil(float64(s.usedCapacity(snapshot)+s.pendingVReplicas()) / (float64(s.capacity) * 0.5)))
 
-				//// Make sure not to scale down past the last pod with placed vpods
-				//if new < snapshot.lastOrdinal+1 {
-				//	new = snapshot.lastOrdinal + 1
-				//}
-				// Disable scaling down until we have a better story with finalizers
-				// See https://github.com/knative-sandbox/eventing-kafka/issues/412
-
-				if new > scale.Spec.Replicas {
+				// Make sure not to scale down past the last pod with placed vpods
+				if new < snapshot.lastOrdinal+1 {
+					new = snapshot.lastOrdinal + 1
+				}
+
+				if new != scale.Spec.Replicas {
 					scale.Spec.Replicas = new
 					s.logger.Infow("updating adapter replicas", zap.Int32("replicas", scale.Spec.Replicas))
 
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index 7122b8c28..86c6bbebc 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -171,7 +171,7 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 	a.logger.Infow("source added", "name", obj.Name)
 }
 
-func (a *Adapter) Remove(ctx context.Context, obj *v1beta1.KafkaSource) {
+func (a *Adapter) Remove(obj *v1beta1.KafkaSource) {
 	a.sourcesMu.Lock()
 	defer a.sourcesMu.Unlock()
 	a.logger.Infow("removing source", "name", obj.Name)
diff --git a/pkg/source/mtadapter/controller.go b/pkg/source/mtadapter/controller.go
index 0b13ea335..2f7654087 100644
--- a/pkg/source/mtadapter/controller.go
+++ b/pkg/source/mtadapter/controller.go
@@ -18,7 +18,8 @@ package mtadapter
 
 import (
 	"context"
-	"strings"
+
+	"k8s.io/client-go/tools/cache"
 
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
@@ -36,7 +37,7 @@ type MTAdapter interface {
 	Update(ctx context.Context, source *v1beta1.KafkaSource)
 
 	// Remove is called when the source has been deleted.
-	Remove(ctx context.Context, source *v1beta1.KafkaSource)
+	Remove(source *v1beta1.KafkaSource)
 }
 
 // NewController initializes the controller and
@@ -51,17 +52,20 @@ func NewController(ctx context.Context, adapter adapter.Adapter) *controller.Imp
 		mtadapter: mtadapter,
 	}
 
-	podName := mtadapter.config.PodName
 	impl := kafkasourcereconciler.NewImpl(ctx, r, func(impl *controller.Impl) controller.Options {
 		return controller.Options{
 			SkipStatusUpdates: true,
-			FinalizerName:     "kafkasources.sources.knative.dev." + podName[strings.LastIndex(podName, "-")+1:],
 		}
 	})
 
 	logging.FromContext(ctx).Info("Setting up event handlers")
 	kafkasourceInformer := kakfasourceinformer.Get(ctx)
-	kafkasourceInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+	kafkasourceInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    impl.Enqueue,
+			UpdateFunc: controller.PassNew(impl.Enqueue),
+			DeleteFunc: r.deleteFunc,
+		})
 
 	return impl
 }
diff --git a/pkg/source/mtadapter/kafkasource.go b/pkg/source/mtadapter/kafkasource.go
index cd63f4616..336420823 100644
--- a/pkg/source/mtadapter/kafkasource.go
+++ b/pkg/source/mtadapter/kafkasource.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 
+	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/reconciler"
 
 	"knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
@@ -34,9 +35,6 @@ type Reconciler struct {
 // Check that our Reconciler implements ReconcileKind.
 var _ kafkasourcereconciler.Interface = (*Reconciler)(nil)
 
-// Check that our Reconciler implements FinalizeKind.
-var _ kafkasourcereconciler.Finalizer = (*Reconciler)(nil)
-
 func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1beta1.KafkaSource) reconciler.Event {
 	if !source.Status.IsReady() {
 		return fmt.Errorf("warning: KafkaSource is not ready")
@@ -48,9 +46,17 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1beta1.KafkaSou
 	return nil
 }
 
-func (r *Reconciler) FinalizeKind(ctx context.Context, source *v1beta1.KafkaSource) reconciler.Event {
-	// Update the adapter state
-	r.mtadapter.Remove(ctx, source)
-
-	return nil
+func (r *Reconciler) deleteFunc(obj interface{}) {
+	if obj == nil {
+		return
+	}
+	acc, err := kmeta.DeletionHandlingAccessor(obj)
+	if err != nil {
+		return
+	}
+	source, ok := acc.(*v1beta1.KafkaSource)
+	if !ok || source == nil {
+		return
+	}
+	r.mtadapter.Remove(source)
 }
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
index c09035a6d..e4b08753f 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/controller.go
@@ -117,6 +117,9 @@ func NewImpl(ctx context.Context, r Interface, classValue string, optionsFns ...
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
index d0043b784..606dc0af7 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go
@@ -173,6 +173,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
index 65771bffa..84b5244e4 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/controller.go
@@ -117,6 +117,9 @@ func NewImpl(ctx context.Context, r Interface, classValue string, optionsFns ...
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
index 1f73a4a43..bfbf932ad 100644
--- a/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
+++ b/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/broker/reconciler.go
@@ -173,6 +173,9 @@ func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versio
 		if opts.SkipStatusUpdates {
 			rec.skipStatusUpdates = true
 		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
index 06d436648..78ff96730 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_controller.go
@@ -271,6 +271,9 @@ func NewImpl(ctx {{.contextContext|raw}}, r Interface{{if .hasClass}}, classValu
 			rec.skipStatusUpdates = true
 		}
 		{{- end}}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	rec.Recorder = createRecorder(ctx, agentName)
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
index 10149fadd..cc6757da8 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
@@ -363,6 +363,9 @@ func NewReconciler(ctx {{.contextContext|raw}}, logger *{{.zapSugaredLogger|raw}
 			rec.skipStatusUpdates = true
 		}
 		{{- end}}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
 	}
 
 	return rec
diff --git a/vendor/knative.dev/pkg/controller/controller.go b/vendor/knative.dev/pkg/controller/controller.go
index 27fe75e40..2a28497fa 100644
--- a/vendor/knative.dev/pkg/controller/controller.go
+++ b/vendor/knative.dev/pkg/controller/controller.go
@@ -152,6 +152,13 @@ func FilterControllerGK(gk schema.GroupKind) func(obj interface{}) bool {
 	}
 }
 
+// FilterController makes it simple to create FilterFunc's for use with
+// cache.FilteringResourceEventHandler that filter based on the
+// controlling resource.
+func FilterController(r kmeta.OwnerRefable) func(obj interface{}) bool {
+	return FilterControllerGK(r.GetGroupVersionKind().GroupKind())
+}
+
 // FilterWithName makes it simple to create FilterFunc's for use with
 // cache.FilteringResourceEventHandler that filter based on a name.
 func FilterWithName(name string) func(obj interface{}) bool {
diff --git a/vendor/knative.dev/pkg/controller/options.go b/vendor/knative.dev/pkg/controller/options.go
index b40b5a359..8d61835e6 100644
--- a/vendor/knative.dev/pkg/controller/options.go
+++ b/vendor/knative.dev/pkg/controller/options.go
@@ -35,6 +35,9 @@ type Options struct {
 	// SkipStatusUpdates configures this reconciler to either do automated status
 	// updates (default) or skip them if this is set to true.
 	SkipStatusUpdates bool
+
+	// DemoteFunc configures the demote function this reconciler uses
+	DemoteFunc func(b reconciler.Bucket)
 }
 
 // OptionsFn is a callback method signature that accepts an Impl and returns
diff --git a/vendor/knative.dev/pkg/metrics/exporter.go b/vendor/knative.dev/pkg/metrics/exporter.go
index b98a18de1..3fc4966fd 100644
--- a/vendor/knative.dev/pkg/metrics/exporter.go
+++ b/vendor/knative.dev/pkg/metrics/exporter.go
@@ -21,7 +21,6 @@ import (
 	"errors"
 	"fmt"
 	"strings"
-	"sync"
 
 	"go.opencensus.io/resource"
 	"go.opencensus.io/stats/view"
@@ -32,9 +31,14 @@ import (
 var (
 	curMetricsExporter view.Exporter
 	curMetricsConfig   *metricsConfig
-	metricsMux         sync.RWMutex
+	mWorker            *metricsWorker
 )
 
+func init() {
+	mWorker = newMetricsWorker()
+	go mWorker.start()
+}
+
 // SecretFetcher is a function (extracted from SecretNamespaceLister) for fetching
 // a specific Secret. This avoids requiring global or namespace list in controllers.
 type SecretFetcher func(string) (*corev1.Secret, error)
@@ -154,28 +158,14 @@ func UpdateExporter(ctx context.Context, ops ExporterOptions, logger *zap.Sugare
 
 	// Updating the metrics config and the metrics exporters needs to be atomic to
 	// avoid using an outdated metrics config with new exporters.
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-
-	if isNewExporterRequired(newConfig) {
-		logger.Info("Flushing the existing exporter before setting up the new exporter.")
-		flushGivenExporter(curMetricsExporter)
-		e, f, err := newMetricsExporter(newConfig, logger)
-		if err != nil {
-			logger.Errorw("Failed to update a new metrics exporter based on metric config", zap.Error(err), "config", newConfig)
-			return err
-		}
-		existingConfig := curMetricsConfig
-		curMetricsExporter = e
-		if err := setFactory(f); err != nil {
-			logger.Errorw("Failed to update metrics factory when loading metric config", zap.Error(err), "config", newConfig)
-			return err
-		}
-		logger.Infof("Successfully updated the metrics exporter; old config: %v; new config %v", existingConfig, newConfig)
+	updateCmd := &updateMetricsConfigWithExporter{
+		ctx:       ctx,
+		newConfig: newConfig,
+		done:      make(chan error),
 	}
-
-	setCurMetricsConfigUnlocked(newConfig)
-	return nil
+	mWorker.c <- updateCmd
+	err = <-updateCmd.done
+	return err
 }
 
 // isNewExporterRequired compares the non-nil newConfig against curMetricsConfig. When backend changes,
@@ -228,27 +218,35 @@ func newMetricsExporter(config *metricsConfig, logger *zap.SugaredLogger) (view.
 }
 
 func getCurMetricsExporter() view.Exporter {
-	metricsMux.RLock()
-	defer metricsMux.RUnlock()
-	return curMetricsExporter
+	readCmd := &readExporter{done: make(chan *view.Exporter)}
+	mWorker.c <- readCmd
+	e := <-readCmd.done
+	return *e
 }
 
 func setCurMetricsExporter(e view.Exporter) {
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-	curMetricsExporter = e
+	setCmd := &setExporter{
+		newExporter: &e,
+		done:        make(chan struct{}),
+	}
+	mWorker.c <- setCmd
+	<-setCmd.done
 }
 
 func getCurMetricsConfig() *metricsConfig {
-	metricsMux.RLock()
-	defer metricsMux.RUnlock()
-	return curMetricsConfig
+	readCmd := &readMetricsConfig{done: make(chan *metricsConfig)}
+	mWorker.c <- readCmd
+	cfg := <-readCmd.done
+	return cfg
 }
 
 func setCurMetricsConfig(c *metricsConfig) {
-	metricsMux.Lock()
-	defer metricsMux.Unlock()
-	setCurMetricsConfigUnlocked(c)
+	setCmd := &setMetricsConfig{
+		newConfig: c,
+		done:      make(chan struct{}),
+	}
+	mWorker.c <- setCmd
+	<-setCmd.done
 }
 
 func setCurMetricsConfigUnlocked(c *metricsConfig) {
diff --git a/vendor/knative.dev/pkg/metrics/metrics_worker.go b/vendor/knative.dev/pkg/metrics/metrics_worker.go
new file mode 100644
index 000000000..0b7753acb
--- /dev/null
+++ b/vendor/knative.dev/pkg/metrics/metrics_worker.go
@@ -0,0 +1,111 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"context"
+
+	"go.opencensus.io/stats/view"
+	"go.uber.org/zap"
+	"knative.dev/pkg/logging"
+)
+
+type metricsWorker struct {
+	c chan command
+}
+
+func newMetricsWorker() *metricsWorker {
+	return &metricsWorker{c: make(chan command)}
+}
+
+type command interface {
+	handleCommand(w *metricsWorker)
+}
+
+type readExporter struct {
+	done chan *view.Exporter
+}
+
+type setExporter struct {
+	newExporter *view.Exporter
+	done        chan struct{}
+}
+
+type readMetricsConfig struct {
+	done chan *metricsConfig
+}
+
+type updateMetricsConfigWithExporter struct {
+	ctx       context.Context
+	newConfig *metricsConfig
+	done      chan error
+}
+
+type setMetricsConfig struct {
+	newConfig *metricsConfig
+	done      chan struct{}
+}
+
+func (cmd *readMetricsConfig) handleCommand(w *metricsWorker) {
+	cmd.done <- curMetricsConfig
+}
+
+func (cmd *setMetricsConfig) handleCommand(w *metricsWorker) {
+	setCurMetricsConfigUnlocked(cmd.newConfig)
+	cmd.done <- struct{}{}
+}
+
+func (cmd *updateMetricsConfigWithExporter) handleCommand(w *metricsWorker) {
+	ctx := cmd.ctx
+	logger := logging.FromContext(ctx)
+	if isNewExporterRequired(cmd.newConfig) {
+		logger.Info("Flushing the existing exporter before setting up the new exporter.")
+		flushGivenExporter(curMetricsExporter)
+		e, f, err := newMetricsExporter(cmd.newConfig, logger)
+		if err != nil {
+			logger.Errorw("Failed to update a new metrics exporter based on metric config", zap.Error(err), "config", cmd.newConfig)
+			cmd.done <- err
+			return
+		}
+		existingConfig := curMetricsConfig
+		curMetricsExporter = e
+		if err := setFactory(f); err != nil {
+			logger.Errorw("Failed to update metrics factory when loading metric config", zap.Error(err), "config", cmd.newConfig)
+			cmd.done <- err
+			return
+		}
+		logger.Infof("Successfully updated the metrics exporter; old config: %v; new config %v", existingConfig, cmd.newConfig)
+	}
+	setCurMetricsConfigUnlocked(cmd.newConfig)
+	cmd.done <- nil
+}
+
+func (w *metricsWorker) start() {
+	for {
+		cmd := <-w.c
+		cmd.handleCommand(w)
+	}
+}
+
+func (cmd *setExporter) handleCommand(w *metricsWorker) {
+	curMetricsExporter = *cmd.newExporter
+	cmd.done <- struct{}{}
+}
+
+func (cmd *readExporter) handleCommand(w *metricsWorker) {
+	cmd.done <- &curMetricsExporter
+}
diff --git a/vendor/knative.dev/pkg/metrics/resource_view.go b/vendor/knative.dev/pkg/metrics/resource_view.go
index 7135af70a..236fd588b 100644
--- a/vendor/knative.dev/pkg/metrics/resource_view.go
+++ b/vendor/knative.dev/pkg/metrics/resource_view.go
@@ -82,10 +82,16 @@ func cleanup() {
 	expiryCutoff := allMeters.clock.Now().Add(-1 * maxMeterExporterAge)
 	allMeters.lock.Lock()
 	defer allMeters.lock.Unlock()
+	resourceViews.lock.Lock()
+	defer resourceViews.lock.Unlock()
 	for key, meter := range allMeters.meters {
 		if key != "" && meter.t.Before(expiryCutoff) {
 			flushGivenExporter(meter.e)
+			// Make a copy of views to avoid data races
+			viewsCopy := copyViews(resourceViews.views)
+			meter.m.Unregister(viewsCopy...)
 			delete(allMeters.meters, key)
+			meter.m.Stop()
 		}
 	}
 }
@@ -139,7 +145,7 @@ func RegisterResourceView(views ...*view.View) error {
 	return nil
 }
 
-// UnregisterResourceView is similar to view.Unregiste(), except that it will
+// UnregisterResourceView is similar to view.Unregister(), except that it will
 // unregister the view across all Resources tracked byt he system, rather than
 // simply the default view.
 func UnregisterResourceView(views ...*view.View) {
diff --git a/vendor/knative.dev/pkg/network/handlers/drain.go b/vendor/knative.dev/pkg/network/handlers/drain.go
index 0ed37ba6d..b8dc6ff1b 100644
--- a/vendor/knative.dev/pkg/network/handlers/drain.go
+++ b/vendor/knative.dev/pkg/network/handlers/drain.go
@@ -17,6 +17,7 @@ limitations under the License.
 package handlers
 
 import (
+	"fmt"
 	"net/http"
 	"sync"
 	"time"
@@ -90,11 +91,11 @@ func (d *Drainer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
-	if network.IsKProbe(r) {
+	if isKProbe(r) {
 		if d.draining() {
 			http.Error(w, "shutting down", http.StatusServiceUnavailable)
 		} else {
-			network.ServeKProbe(w, r)
+			serveKProbe(w, r)
 		}
 		return
 	}
@@ -146,3 +147,21 @@ func (d *Drainer) draining() bool {
 	defer d.RUnlock()
 	return d.timer != nil
 }
+
+// isKProbe returns true if the request is a knatvie probe.
+func isKProbe(r *http.Request) bool {
+	return r.Header.Get(network.ProbeHeaderName) == network.ProbeHeaderValue
+}
+
+// serveKProbe serve KProbe requests.
+func serveKProbe(w http.ResponseWriter, r *http.Request) {
+	hh := r.Header.Get(network.HashHeaderName)
+	if hh == "" {
+		http.Error(w,
+			fmt.Sprintf("a probe request must contain a non-empty %q header", network.HashHeaderName),
+			http.StatusBadRequest)
+		return
+	}
+	w.Header().Set(network.HashHeaderName, hh)
+	w.WriteHeader(http.StatusOK)
+}
diff --git a/vendor/knative.dev/pkg/network/network.go b/vendor/knative.dev/pkg/network/network.go
index b2ac63ce7..cf5feaa8f 100644
--- a/vendor/knative.dev/pkg/network/network.go
+++ b/vendor/knative.dev/pkg/network/network.go
@@ -17,7 +17,6 @@ limitations under the License.
 package network
 
 import (
-	"fmt"
 	"net/http"
 	"strings"
 	"time"
@@ -52,33 +51,19 @@ const (
 	// uses to find out which version of the networking config is deployed.
 	HashHeaderName = "K-Network-Hash"
 
+	// KubeProbeUAPrefix is the prefix for the User-Agent header.
 	// Since K8s 1.8, prober requests have
 	//   User-Agent = "kube-probe/{major-version}.{minor-version}".
 	KubeProbeUAPrefix = "kube-probe/"
 
+	// KubeletProbeHeaderName is the header name to augment the probes, because
 	// Istio with mTLS rewrites probes, but their probes pass a different
-	// user-agent.  So we augment the probes with this header.
+	// user-agent.
 	KubeletProbeHeaderName = "K-Kubelet-Probe"
 )
 
 // IsKubeletProbe returns true if the request is a Kubernetes probe.
 func IsKubeletProbe(r *http.Request) bool {
-	return strings.HasPrefix(r.Header.Get("User-Agent"), KubeProbeUAPrefix) ||
+	return strings.HasPrefix(r.Header.Get(UserAgentKey), KubeProbeUAPrefix) ||
 		r.Header.Get(KubeletProbeHeaderName) != ""
 }
-
-// IsKProbe returns true if the request is a knatvie probe.
-func IsKProbe(r *http.Request) bool {
-	return r.Header.Get(ProbeHeaderName) == ProbeHeaderValue
-}
-
-// ServeKProbe serve KProbe requests.
-func ServeKProbe(w http.ResponseWriter, r *http.Request) {
-	hh := r.Header.Get(HashHeaderName)
-	if hh == "" {
-		http.Error(w, fmt.Sprintf("a probe request must contain a non-empty %q header", HashHeaderName), http.StatusBadRequest)
-		return
-	}
-	w.Header().Set(HashHeaderName, hh)
-	w.WriteHeader(http.StatusOK)
-}
diff --git a/vendor/knative.dev/pkg/source/source_stats_reporter.go b/vendor/knative.dev/pkg/source/source_stats_reporter.go
index b6d1ca93b..f343d9d75 100644
--- a/vendor/knative.dev/pkg/source/source_stats_reporter.go
+++ b/vendor/knative.dev/pkg/source/source_stats_reporter.go
@@ -35,6 +35,12 @@ var (
 		stats.UnitDimensionless,
 	)
 
+	// retryEventCountM is a counter which records the number of events sent by the source in retries.
+	retryEventCountM = stats.Int64(
+		"retry_event_count",
+		"Number of retry events sent",
+		stats.UnitDimensionless,
+	)
 	// Create the tag keys that will be used to add tags to our measurements.
 	// Tag keys must conform to the restrictions described in
 	// go.opencensus.io/tag/validate.go. Currently those restrictions are:
@@ -51,6 +57,7 @@ var (
 	responseTimeout        = tag.MustNewKey(metricskey.LabelResponseTimeout)
 )
 
+// ReportArgs defines the arguments for reporting metrics.
 type ReportArgs struct {
 	Namespace     string
 	EventType     string
@@ -69,6 +76,7 @@ func init() {
 type StatsReporter interface {
 	// ReportEventCount captures the event count. It records one per call.
 	ReportEventCount(args *ReportArgs, responseCode int) error
+	ReportRetryEventCount(args *ReportArgs, responseCode int) error
 }
 
 var _ StatsReporter = (*reporter)(nil)
@@ -98,6 +106,15 @@ func (r *reporter) ReportEventCount(args *ReportArgs, responseCode int) error {
 	return nil
 }
 
+func (r *reporter) ReportRetryEventCount(args *ReportArgs, responseCode int) error {
+	ctx, err := r.generateTag(args, responseCode)
+	if err != nil {
+		return err
+	}
+	metrics.Record(ctx, retryEventCountM.M(1))
+	return nil
+}
+
 func (r *reporter) generateTag(args *ReportArgs, responseCode int) (context.Context, error) {
 	return tag.New(
 		r.ctx,
@@ -132,6 +149,12 @@ func register() {
 			Aggregation: view.Count(),
 			TagKeys:     tagKeys,
 		},
+		&view.View{
+			Description: retryEventCountM.Description(),
+			Measure:     retryEventCountM,
+			Aggregation: view.Count(),
+			TagKeys:     tagKeys,
+		},
 	); err != nil {
 		panic(err)
 	}
diff --git a/vendor/knative.dev/pkg/system/clock.go b/vendor/knative.dev/pkg/system/clock.go
deleted file mode 100644
index 7d99d9b5c..000000000
--- a/vendor/knative.dev/pkg/system/clock.go
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
-Copyright 2019 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package system
-
-import (
-	"time"
-)
-
-// Mockable interface for time based testing
-type Clock interface {
-	Now() time.Time
-}
-
-type RealClock struct{}
-
-func (RealClock) Now() time.Time {
-	return time.Now()
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index d6e00cb59..886acaa5d 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.21.0
+# knative.dev/eventing v0.21.1-0.20210302004649-f22066824960
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1219,7 +1219,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-20210216013737-584933f8280b
+# knative.dev/pkg v0.0.0-20210226182947-9039dc189ced
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
index f8dd508e7..e1a376c95 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -252,6 +252,95 @@ func TestAuthFromSarama(t *testing.T) {
 	}
 }
 
+func TestStringifyHeaders(t *testing.T) {
+
+	// Define The TestCase Struct
+	type TestCase struct {
+		name     string
+		headers  []sarama.RecordHeader
+		expected map[string][]string
+	}
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{
+			name:     "Nil Slice",
+			headers:  nil,
+			expected: map[string][]string{},
+		},
+		{
+			name:     "Empty Slice",
+			headers:  []sarama.RecordHeader{},
+			expected: map[string][]string{},
+		},
+		{
+			name: "One Header",
+			headers: []sarama.RecordHeader{
+				{
+					Key:   []byte("one-key"),
+					Value: []byte("one-value"),
+				},
+			},
+			expected: map[string][]string{
+				"one-key": {"one-value"},
+			},
+		},
+		{
+			name: "Two Different Headers",
+			headers: []sarama.RecordHeader{
+				{
+					Key:   []byte("one-key"),
+					Value: []byte("one-value"),
+				},
+				{
+					Key:   []byte("two-key"),
+					Value: []byte("two-value"),
+				},
+			},
+			expected: map[string][]string{
+				"one-key": {"one-value"},
+				"two-key": {"two-value"},
+			},
+		},
+		{
+			name: "Two Identical Headers",
+			headers: []sarama.RecordHeader{
+				{
+					Key:   []byte("one-key"),
+					Value: []byte("one-value"),
+				},
+				{
+					Key:   []byte("one-key"),
+					Value: []byte("two-value"),
+				},
+			},
+			expected: map[string][]string{
+				"one-key": {"one-value", "two-value"},
+			},
+		},
+	}
+
+	// Run The TestCases
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+
+			// Perform The Test
+			stringHeaders := StringifyHeaders(testCase.headers)
+
+			// Perform The Test With Pointers (should have same output)
+			ptrs := make([]*sarama.RecordHeader, len(testCase.headers))
+			for index, header := range testCase.headers {
+				ptrs[index] = &sarama.RecordHeader{Key: header.Key, Value: header.Value}
+			}
+			stringHeadersFromPtrs := StringifyHeaderPtrs(ptrs)
+
+			// Verify Expected State
+			assert.Equal(t, testCase.expected, stringHeaders)
+			assert.Equal(t, testCase.expected, stringHeadersFromPtrs)
+		})
+	}
+}
+
 func verifyTestEKConfigSettings(t *testing.T, saramaConfig *sarama.Config, eventingKafkaConfig *commonconfig.EventingKafkaConfig) {
 	// Quick checks to make sure the loaded configs aren't complete junk
 	assert.Equal(t, commontesting.OldUsername, saramaConfig.Net.SASL.User)
diff --git a/pkg/common/client/config_test.go b/pkg/common/client/config_test.go
index 2d4ac0be1..b4009f881 100644
--- a/pkg/common/client/config_test.go
+++ b/pkg/common/client/config_test.go
@@ -60,6 +60,25 @@ Consumer:
 `
 )
 
+const EKDefaultSaramaConfigWithoutCerts = `
+Net:
+  TLS:
+    Enable: true
+    Config:
+  SASL:
+    Mechanism: PLAIN
+    Version: 1
+Metadata:
+  RefreshFrequency: 300000000000
+Consumer:
+  Offsets:
+    AutoCommit:
+        Interval: 5000000000
+    Retention: 604800000000000
+  Return:
+    Errors: true
+`
+
 const EKDefaultSaramaConfigWithRootCert = `
 Net:
   TLS:
@@ -116,6 +135,97 @@ Consumer:
     Errors: true
 `
 
+const EKDefaultSaramaConfigWithTwoRootCerts = `
+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-----
+        -----BEGIN CERTIFICATE-----
+        MIIGDzCCA/egAwIBAgIUWq6j7u/25wPQiNMPZqL6Vy0rkvQwDQYJKoZIhvcNAQEL
+        BQAwgY4xCzAJBgNVBAYTAkRFMRswGQYDVQQIDBJCYWRlbi1XdWVydHRlbWJlcmcx
+        ETAPBgNVBAcMCFdhbGxkb3JmMQ8wDQYDVQQKDAZTQVAgU0UxHzAdBgNVBAsMFlNB
+        UCBDUCBEYXRhIE1hbmFnZW1lbnQxHTAbBgNVBAMMFFNBUCBTRSBLYWZrYSBSb290
+        IENBMB4XDTIxMDIxNzE0MjgyOVoXDTI0MDYwMTE0MjgyOVowgY4xCzAJBgNVBAYT
+        AkRFMRswGQYDVQQIDBJCYWRlbi1XdWVydHRlbWJlcmcxETAPBgNVBAcMCFdhbGxk
+        b3JmMQ8wDQYDVQQKDAZTQVAgU0UxHzAdBgNVBAsMFlNBUCBDUCBEYXRhIE1hbmFn
+        ZW1lbnQxHTAbBgNVBAMMFFNBUCBTRSBLYWZrYSBSb290IENBMIICIjANBgkqhkiG
+        9w0BAQEFAAOCAg8AMIICCgKCAgEAoLH563sTndme/L5sUH/LggOO5sgRk6D6eYJi
+        YgPZFX/o8a0dctgMTBK574kPAZeniwvOIUUeGhKM/Ho4ou/B3XDD/8WtoOIZt4lj
+        QGyeSlZVFtCI3Ip/ENj5gIuFhO6N209mpw9wIgjhCdCTU+g6uaAe440aqCMhVO5I
+        54bHp4hSHwkLdNZnsRjSopufEKxWbJm52w7lKHwBwNc0cAckiagne6FbBTliF4K/
+        bgmfRRwP9fzjcmgno+GGgKastNkJVh6Ku/2NCpOKSzQiGNAQAvqbZbpQIkCt4/PK
+        Nj4c5pjbRSajwt8JWNHRJ9muJz2Db7ZceQF7H/vcAwlu/IK/ARHygfA72IY164du
+        QN1qJyi1DkPs1qm5cuA8WbfL8KSqIFeiKoInHL524zeK2Yt5b89sS00ydtLSYbri
+        1kd31McUCHcj66DDX5xBbNg7bgRx09DNpfwKlUnwC7tphz1DLJprJ9OvhjEZi0w7
+        TNrL21yETIKGG0qSgebIhLPWTG650i2hEM2dF1t9RfVB4qdhaxS194rkwQPZ2lQ8
+        74hWnuqiZ3L4XhMp1AO86UoFEWyeCbIFsZ4YHnCNfBOO4t8LSPpUiV9bTIwqG549
+        upI9tWMZMVV59GiuA4M9oit7vYa1eNxpUhKWpqAVCA66tETln04umAKsywt06pUn
+        0udI1OsCAwEAAaNjMGEwHQYDVR0OBBYEFGRMbSqCPP1qeNX3hEYn4YAClp/cMB8G
+        A1UdIwQYMBaAFGRMbSqCPP1qeNX3hEYn4YAClp/cMA8GA1UdEwEB/wQFMAMBAf8w
+        DgYDVR0PAQH/BAQDAgGGMA0GCSqGSIb3DQEBCwUAA4ICAQBRNkbWuq13OZVAC+KG
+        tjQJWkAym80St+5Ta2XiethvjnBi4TJGP8pjmMJhbUcQ6HNHQlHSFUDE5RKZkZvl
+        EggAwIhKv49MuguiDUhe2aguQvcUV3VHj96/l6ryHsMz4tT6zQBhEaw2Ejgf6zDO
+        RHBMh2AiEo8CDGaS9TOpSQQSmZf3xWdOAagjxbMKWYjbU2EMQV2vA0+lv2RGtZGW
+        WUMaj9kToYCYqFsWxVBQd8EQ/9lIVrEyi9raI5X06ofiKlpG93oEtiL8SQnp27Wa
+        ynQVBA0iS5BD34uXLGT7K6QnijR+FFKWeTFTCCKtiVrD5S3+zs29d2stQ9pvPo8x
+        hfUV76mik+i3XX6/rpyoCtOHNHW/38LPMHzwyQQ1DgJ3TJ/0WHuWZvtzFIUwl5Ie
+        4lJ3tfrh5sXyHOMclWM5Sl9TwJvGJIo4XAP+qnFInL8zSN72rzliJdZMbJDHM6Vz
+        sGz3KWmgntiTDg+1hLvBD/D6Le1axC8qg7j/QUL8D/YFVy4AZnhuMYdVJBM5oNs5
+        gec0L0mhPTD8fqVpzr6nZ/K2ZL5xlOSujE5+w9gmjtuGZwVpI7wOmNSgaG6/u7CP
+        31OCuKAlbmDBYuwjaCIOWZL+BBj7bQcI6hZEP4oj+vSWe0xUPLvH2u2Jr26arNEs
+        771uezZAFqd1GLLL8ZYRmCsAMg==
+        -----END CERTIFICATE-----
+  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 TestBuildSaramaConfig(t *testing.T) {
 	logger := logtesting.TestLogger(t)
diff --git a/pkg/common/client/yaml_test.go b/pkg/common/client/yaml_test.go
index 50a33f20c..e51861839 100644
--- a/pkg/common/client/yaml_test.go
+++ b/pkg/common/client/yaml_test.go
@@ -33,137 +33,23 @@ func TestExtractRootCerts(t *testing.T) {
 		expectedRootCertCount int
 	}{
 		{
-			name:       "single cert",
-			yamlBefore: EKDefaultSaramaConfigWithRootCert,
-			expectedYamlAfter: `
-Net:
-  TLS:
-    Enable: true
-    Config:
-  SASL:
-    Mechanism: PLAIN
-    Version: 1
-Metadata:
-  RefreshFrequency: 300000000000
-Consumer:
-  Offsets:
-    AutoCommit:
-        Interval: 5000000000
-    Retention: 604800000000000
-  Return:
-    Errors: true
-`,
+			name:                  "single cert",
+			yamlBefore:            EKDefaultSaramaConfigWithRootCert,
+			expectedYamlAfter:     EKDefaultSaramaConfigWithoutCerts,
 			expectedRootCertCount: 1,
 		},
 		{
-			name: "no cert",
-			yamlBefore: `
-Net:
-  TLS:
-    Enable: true
-    Config:
-`,
-			expectedYamlAfter: `
-Net:
-  TLS:
-    Enable: true
-    Config:
-`,
+			name:                  "no cert",
+			yamlBefore:            EKDefaultSaramaConfigWithoutCerts,
+			expectedYamlAfter:     EKDefaultSaramaConfigWithoutCerts,
 			expectedRootCertCount: 0,
 		},
-		// TODO: need some YAML magic to have the more than one multiline-string-element in the array
-		//		{
-		//			name: "multiple certs",
-		//			yamlBefore: `
-		//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-----
-		//      - |-
-		//		-----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-----
-		//`,
-		//			expectedYamlAfter: `
-		//Net:
-		//  TLS:
-		//    Enable: true
-		//    Config:
-		//  SASL:
-		//    Mechanism: PLAIN
-		//    Version: 1
-		//`,
-		//			expectedRootCertCount: 2,
-		//		},
+		{
+			name:                  "multiple certs",
+			yamlBefore:            EKDefaultSaramaConfigWithTwoRootCerts,
+			expectedYamlAfter:     EKDefaultSaramaConfigWithoutCerts,
+			expectedRootCertCount: 2,
+		},
 	}
 
 	for _, tc := range tt {
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index d0579abb9..492653c5a 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -90,7 +90,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 		t.Error("sub-adapter failed to start after 100 ms")
 	}
 
-	adapter.Remove(ctx, &sourcesv1beta1.KafkaSource{
+	adapter.Remove(&sourcesv1beta1.KafkaSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-name",
 			Namespace: "test-ns",
diff --git a/vendor/knative.dev/pkg/reconciler/testing/clock.go b/vendor/knative.dev/pkg/reconciler/testing/clock.go
deleted file mode 100644
index 44ba77cdb..000000000
--- a/vendor/knative.dev/pkg/reconciler/testing/clock.go
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
-Copyright 2019 The Knative Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package testing
-
-import (
-	"time"
-)
-
-type FakeClock struct {
-	Time time.Time
-}
-
-func (c FakeClock) Now() time.Time {
-	return c.Time
-}