Repository: knative-sandbox/eventing-kafka
Pull request #418 (2021-02-26 18:32:19)

Stringify the sarama RecordHeader slices for easier log reading
This will change the way the debug log output formats messages for the distributed dispatcher and receiver.  They used to be at the whim of zap.Any() which produced a lot of base64 strings for the headers and the message, which is obnoxious to decode when looking through a log and provided no benefit.

## Proposed Changes

- 🧽 Added helper functions to convert []sarama.RecordHeader and []*sarama.RecordHeader inputs into basic string maps so that the logs of these are cleaner to human eyes, and called these from the ProduceKafkaMessage and consumeMessage functions, respectively.

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 **aliok** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @aliok` 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":["aliok"]} -->
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/418?src=pr&el=h1) Report
> Merging [#418](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/418?src=pr&el=desc) (452638c) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/c7ab439c66df5641d312c59731a8d33ac0bb8b1a?el=desc) (c7ab439) will **decrease** coverage by `0.08%`.
> The diff coverage is `100.00%`.

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

```diff
@@            Coverage Diff             @@
##           master     #418      +/-   ##
==========================================
- Coverage   73.13%   73.05%   -0.09%     
==========================================
  Files         129      129              
  Lines        4973     4984      +11     
==========================================
+ Hits         3637     3641       +4     
- Misses       1100     1107       +7     
  Partials      236      236              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/418?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/channel/distributed/common/util/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/418/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL3V0aWwvdXRpbC5nbw==) | `100.00% <100.00%> (ø)` | |
| [...annel/distributed/dispatcher/dispatcher/handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/418/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2hhbmRsZXIuZ28=) | `87.23% <100.00%> (ø)` | |
| [.../channel/distributed/receiver/producer/producer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/418/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvcmVjZWl2ZXIvcHJvZHVjZXIvcHJvZHVjZXIuZ28=) | `69.56% <100.00%> (+0.81%)` | :arrow_up: |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/418/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `71.20% <0.00%> (-5.60%)` | :arrow_down: |

------

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

travis-minke-sap: /ok-to-test
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/receiver/producer/producer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-go-coverage/1365381961471758336/artifacts/line-cov.html#file2) | 77.9% | 78.1% | 0.2

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 7a502e531ad7afb97a9a229245e1907f04e73616 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-unit-tests/1365390334883794944) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/common/kafka/sarama/sarama.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-go-coverage/1365390334904766464/artifacts/line-cov.html#file0) | 89.3% | 92.1% | 2.8

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 7a502e531ad7afb97a9a229245e1907f04e73616 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-unit-tests/1365390334883794944) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | 7a502e531ad7afb97a9a229245e1907f04e73616 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-build-tests/1365390334854434816) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/common/kafka/sarama/sarama.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-go-coverage/1365392881057009664/artifacts/line-cov.html#file0) | 89.3% | 92.1% | 2.8
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-go-coverage/1365392881057009664/artifacts/line-cov.html#file1) | 91.4% | 88.6% | -2.9
[pkg/channel/distributed/receiver/producer/producer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/418/pull-knative-sandbox-eventing-kafka-go-coverage/1365392881057009664/artifacts/line-cov.html#file2) | 77.9% | 78.1% | 0.2


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