Repository: knative-sandbox/eventing-kafka
Pull request #417 (2021-02-26 18:27:10)

Fix distributed multi cert tests
This change is only to the TestExtractRootCerts function as I needed the multi-cert functionality and wanted it to be testable. 

## Proposed Changes

- 🧽  Update TestExtractRootCerts test to include multiple certificates 


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 **grantr** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @grantr` in a comment when ready.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["grantr"]} -->
knative-prow-robot: Hi @eric-sap. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<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.
</details>

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417?src=pr&el=h1) Report
> Merging [#417](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417?src=pr&el=desc) (4c55e6c) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/63ceba9cec9e8c070eb1e639897177a145c1f6b7?el=desc) (63ceba9) will **increase** coverage by `0.18%`.
> The diff coverage is `42.85%`.

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

```diff
@@            Coverage Diff             @@
##           master     #417      +/-   ##
==========================================
+ Coverage   72.95%   73.13%   +0.18%     
==========================================
  Files         129      129              
  Lines        5002     4973      -29     
==========================================
- Hits         3649     3637      -12     
+ Misses       1116     1100      -16     
+ Partials      237      236       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/channel/distributed/common/k8s/context.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2s4cy9jb250ZXh0Lmdv) | `57.14% <ø> (+9.52%)` | :arrow_up: |
| [...kg/channel/distributed/receiver/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvcmVjZWl2ZXIvY2hhbm5lbC9jaGFubmVsLmdv) | `77.41% <ø> (+16.44%)` | :arrow_up: |
| [pkg/common/scheduler/statefulset/autoscaler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvYXV0b3NjYWxlci5nbw==) | `62.22% <0.00%> (-1.61%)` | :arrow_down: |
| [pkg/source/mtadapter/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvY29udHJvbGxlci5nbw==) | `0.00% <0.00%> (ø)` | |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `70.11% <55.55%> (-0.26%)` | :arrow_down: |
| [...annel/distributed/dispatcher/dispatcher/handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2hhbmRsZXIuZ28=) | `87.23% <100.00%> (-3.54%)` | :arrow_down: |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/417/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `76.80% <0.00%> (+5.59%)` | :arrow_up: |

------

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

travis-minke-sap: /okt-to-test

Patch:
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/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)
 }


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",