Repository: knative-sandbox/eventing-kafka
Pull request #410 (2021-02-19 14:34:47)

Refactor distributed component mains to handle Kubernetes config better
Refactoring distributed component `main`s to make them use more of the functionality in sharedmain.

I wasn't able to [debug](https://github.com/knative/eventing/blob/master/DEVELOPMENT.md#debugging-knative-controllers-and-friends-locally-with-intellij-idea) the receiver and dispatcher locally because they required me to provide additional stuff like Kube URL etc. Once I provided these, this time the debugging failed because of missing service account token.

Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Use `ParseAndGetRESTConfigOrDie` of `knative/pkg` which handles in-cluster or out-of-cluster situations better than `BuildConfigFromFlags` of the same package
- Change above also made these `main`s more similar to other Knative controllers with injection (more effort needed in this area)

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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-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/k8s/context.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-go-coverage/1362772636421066752/artifacts/line-cov.html#file0) | 60.0% | 71.4% | 11.4
[pkg/channel/distributed/receiver/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-go-coverage/1362772636421066752/artifacts/line-cov.html#file1) | 71.8% | 86.7% | 14.9

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

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

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 5078067cb9f379ace764d1e0d64c0039c145dc97 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-unit-tests/1362772636395900928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | 5078067cb9f379ace764d1e0d64c0039c145dc97 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-build-tests/1362772636370735104) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 5078067cb9f379ace764d1e0d64c0039c145dc97 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-unit-tests/1362772636395900928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | 5078067cb9f379ace764d1e0d64c0039c145dc97 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-build-tests/1362772636370735104) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 5078067cb9f379ace764d1e0d64c0039c145dc97 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1362772636320403456) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/410?src=pr&el=h1) Report
> Merging [#410](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/410?src=pr&el=desc) (9ab3386) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/6b130f0198e44a9ec44c2eae21371bc47272cc82?el=desc) (6b130f0) will **increase** coverage by `0.11%`.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff             @@
##           master     #410      +/-   ##
==========================================
+ Coverage   73.07%   73.18%   +0.11%     
==========================================
  Files         129      129              
  Lines        5002     4967      -35     
==========================================
- Hits         3655     3635      -20     
+ Misses       1111     1096      -15     
  Partials      236      236              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/410?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/channel/distributed/common/k8s/context.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/410/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/410/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/410/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvYXV0b3NjYWxlci5nbw==) | `63.82% <0.00%> (-12.77%)` | :arrow_down: |
| [...annel/distributed/dispatcher/dispatcher/handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/410/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2hhbmRsZXIuZ28=) | `87.23% <0.00%> (-3.54%)` | :arrow_down: |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/410/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/410?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/410?src=pr&el=footer). Last update [6b130f0...9ab3386](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/410?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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/k8s/context.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-go-coverage/1362777508495233024/artifacts/line-cov.html#file0) | 60.0% | 71.4% | 11.4
[pkg/channel/distributed/receiver/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/410/pull-knative-sandbox-eventing-kafka-go-coverage/1362777508495233024/artifacts/line-cov.html#file1) | 71.8% | 86.7% | 14.9

aliok: /assign @matzew @travis-minke-sap @eric-sap 
knative-prow-robot: @aliok: GitHub didn't allow me to assign the following users: eric-sap.

Note that only [knative-sandbox members](https://github.com/orgs/knative-sandbox/people), repo collaborators and people who have commented on this issue/PR can be assigned. Additionally, issues/PRs can only have 10 assignees at the same time.
For more information please see [the contributor guide](https://git.k8s.io/community/contributors/guide/first-contribution.md#issue-assignment-in-github)

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka/pull/410#issuecomment-782138923):

>/assign @matzew @travis-minke-sap @eric-sap 


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>
knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

<details>

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

Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 568a992e5..b5543eba2 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -18,17 +18,16 @@ package main
 
 import (
 	"context"
-	"flag"
 	"strconv"
 	"strings"
 
 	"knative.dev/eventing/pkg/kncloudevents"
+	"knative.dev/pkg/injection"
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/clientcmd"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	commonk8s "knative.dev/eventing-kafka/pkg/channel/distributed/common/k8s"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
@@ -38,8 +37,9 @@ import (
 	dispatch "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/dispatcher"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/env"
 	dispatcherhealth "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/health"
-	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
+	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	kncontroller "knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	eventingmetrics "knative.dev/pkg/metrics"
@@ -50,18 +50,32 @@ import (
 var (
 	logger     *zap.Logger
 	dispatcher dispatch.Dispatcher
-	serverURL  = flag.String("server", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
-	kubeconfig = flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
 )
 
 // The Main Function (Go Command)
 func main() {
 
-	// Parse The Flags For Local Development Usage
-	flag.Parse()
+	ctx := signals.NewContext()
+
+	// Create The K8S Configuration (In-Cluster By Default / Cmd Line Flags For Out-Of-Cluster Usage)
+	k8sConfig := injection.ParseAndGetRESTConfigOrDie()
+
+	// TODO: do we really need these? I moved them up
+	const numControllers = 1
+	k8sConfig.QPS = numControllers * rest.DefaultQPS
+	k8sConfig.Burst = numControllers * rest.DefaultBurst
+
+	// Put The Kubernetes Config Into The Context Where The Injection Framework Expects It
+	ctx = injection.WithConfig(ctx, k8sConfig)
+
+	// Create A New Kubernetes Client From The K8S Configuration
+	k8sClient := kubernetes.NewForConfigOrDie(k8sConfig)
+
+	// Put The Kubernetes Client Into The Context Where The Injection Framework Expects It
+	ctx = context.WithValue(ctx, injectionclient.Key{}, k8sClient)
 
 	// Initialize A Knative Injection Lite Context (K8S Client & Logger)
-	ctx := commonk8s.LoggingContext(signals.NewContext(), constants.Component, *serverURL, *kubeconfig)
+	ctx = commonk8s.LoggingContext(ctx, constants.Component, k8sClient)
 
 	// Get The Logger From The Context & Defer Flushing Any Buffered Log Entries On Exit
 	logger = logging.FromContext(ctx).Desugar()
@@ -141,18 +155,9 @@ func main() {
 		logger.Fatal("Failed To Start Secret Watcher", zap.Error(err))
 	}
 
-	config, err := clientcmd.BuildConfigFromFlags(*serverURL, *kubeconfig)
-	if err != nil {
-		logger.Fatal("Error building kubeconfig", zap.Error(err))
-	}
-
-	const numControllers = 1
-	config.QPS = numControllers * rest.DefaultQPS
-	config.Burst = numControllers * rest.DefaultBurst
-	kafkaClientSet := versioned.NewForConfigOrDie(config)
-	kubeClient := kubernetes.NewForConfigOrDie(config)
+	kafkaClient := kafkaclientset.NewForConfigOrDie(k8sConfig)
 
-	kafkaInformerFactory := externalversions.NewSharedInformerFactory(kafkaClientSet, environment.ResyncPeriod)
+	kafkaInformerFactory := externalversions.NewSharedInformerFactory(kafkaClient, environment.ResyncPeriod)
 
 	// Create KafkaChannel Informer
 	kafkaChannelInformer := kafkaInformerFactory.Messaging().V1beta1().KafkaChannels()
@@ -164,8 +169,8 @@ func main() {
 			environment.ChannelKey,
 			dispatcher,
 			kafkaChannelInformer,
-			kubeClient,
-			kafkaClientSet,
+			k8sClient,
+			kafkaClient,
 			ctx.Done(),
 		),
 	}
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 65add32e2..262bf0159 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -18,7 +18,6 @@ package main
 
 import (
 	"context"
-	"flag"
 	"net/http"
 	"strconv"
 	"strings"
@@ -27,6 +26,7 @@ import (
 	"github.com/google/uuid"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	commonk8s "knative.dev/eventing-kafka/pkg/channel/distributed/common/k8s"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
@@ -37,28 +37,41 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/env"
 	channelhealth "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/producer"
+	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	eventingchannel "knative.dev/eventing/pkg/channel"
+	injectionclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/injection"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/logging"
 	eventingmetrics "knative.dev/pkg/metrics"
+	"knative.dev/pkg/signals"
 )
 
 // Variables
 var (
 	logger        *zap.Logger
-	serverURL     = flag.String("server", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
-	kubeconfig    = flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
 	kafkaProducer *producer.Producer
 )
 
 // The Main Function (Go Command)
 func main() {
 
-	// Parse The Flags For Local Development Usage
-	flag.Parse()
+	ctx := signals.NewContext()
+
+	// Create The K8S Configuration (In-Cluster By Default / Cmd Line Flags For Out-Of-Cluster Usage)
+	k8sConfig := injection.ParseAndGetRESTConfigOrDie()
+
+	// Put The Kubernetes Config Into The Context Where The Injection Framework Expects It
+	ctx = injection.WithConfig(ctx, k8sConfig)
+
+	// Create A New Kubernetes Client From The K8S Configuration
+	k8sClient := kubernetes.NewForConfigOrDie(k8sConfig)
+
+	// Put The Kubernetes Client Into The Context Where The Injection Framework Expects It
+	ctx = context.WithValue(ctx, injectionclient.Key{}, k8sClient)
 
 	// Initialize A Knative Injection Lite Context (K8S Client & Logger)
-	ctx := commonk8s.LoggingContext(context.Background(), constants.Component, *serverURL, *kubeconfig)
+	ctx = commonk8s.LoggingContext(ctx, constants.Component, k8sClient)
 
 	// Get The Logger From The Context & Defer Flushing Any Buffered Log Entries On Exit
 	logger = logging.FromContext(ctx).Desugar()
@@ -106,8 +119,12 @@ func main() {
 	if err != nil {
 		logger.Fatal("Failed To Initialize Health Server - Terminating", zap.Error(err))
 	}
+
+	// Create A New Kafka Client From The K8S Config
+	kafkaClient := kafkaclientset.NewForConfigOrDie(k8sConfig)
+
 	// Initialize The KafkaChannel Lister Used To Validate Events
-	err = channel.InitializeKafkaChannelLister(ctx, *serverURL, *kubeconfig, healthServer, environment.ResyncPeriod)
+	err = channel.InitializeKafkaChannelLister(ctx, kafkaClient, healthServer, environment.ResyncPeriod)
 	if err != nil {
 		logger.Fatal("Failed To Initialize KafkaChannel Lister", zap.Error(err))
 	}
diff --git a/go.mod b/go.mod
index 004c57ff4..68c2ab47d 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,7 @@ require (
 	k8s.io/apimachinery v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
-	knative.dev/eventing v0.20.1-0.20210217232802-fb8e3b49931a
+	knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
 	knative.dev/pkg v0.0.0-20210216013737-584933f8280b
 )
diff --git a/go.sum b/go.sum
index eb62b783b..f4ac058e3 100644
--- a/go.sum
+++ b/go.sum
@@ -1244,16 +1244,14 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6 h1:+WnxoVtG8TMiudHBSEtrVL
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K8Hf8whTseBgJcg=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.20.1-0.20210217232802-fb8e3b49931a h1:WNI3FopMdy0NwNfVeJqkLQFOlX5U+G5ECuKj3Ih6vJc=
-knative.dev/eventing v0.20.1-0.20210217232802-fb8e3b49931a/go.mod h1:3XibpVIriBFW31rn4uFQlTzqZgyKHgD4CTJHBwEPrUQ=
+knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a h1:3cqlUxlZ6M+EZA80OTroRXz7XNNImc4Qix01jqOA1Bo=
+knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a/go.mod h1:JjbVEOTJJHqo9CTxbTfrMn018hG8fOr3UfBoCJ7KWaA=
 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-20210212203835-448ae657fb5f h1:pz+UDOTsCIClHXGM/SgCzkctLxgi3+P0C26x6Ujlqho=
-knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
 knative.dev/pkg v0.0.0-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-20210215155505-fe22009753f9/go.mod h1:vB3GnWUdlkvtmoVUP9anmh82S1QC/LJKtOrAMA1COtw=
+knative.dev/reconciler-test v0.0.0-20210216030508-77f50054d024/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/k8s/context.go b/pkg/channel/distributed/common/k8s/context.go
index 921334bb2..f63a37547 100644
--- a/pkg/channel/distributed/common/k8s/context.go
+++ b/pkg/channel/distributed/common/k8s/context.go
@@ -24,27 +24,12 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
-	k8sclientcmd "k8s.io/client-go/tools/clientcmd"
-	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	configmap "knative.dev/pkg/configmap/informer"
 	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
 )
 
-// K8sClientWrapper Used To Facilitate Unit Testing
-var K8sClientWrapper = func(serverUrl string, kubeconfigPath string) kubernetes.Interface {
-
-	// Create The K8S Configuration (In-Cluster By Default / Cmd Line Flags For Out-Of-Cluster Usage)
-	k8sConfig, err := k8sclientcmd.BuildConfigFromFlags(serverUrl, kubeconfigPath)
-	if err != nil {
-		log.Fatalf("Failed To Build Kubernetes Config: %v", err)
-	}
-
-	// Create A New Kubernetes Client From The K8S Configuration
-	return kubernetes.NewForConfigOrDie(k8sConfig)
-}
-
 //
 // Initialize The Specified Context With A K8S Client & Logger (ConfigMap Watcher)
 //
@@ -54,14 +39,7 @@ var K8sClientWrapper = func(serverUrl string, kubeconfigPath string) kubernetes.
 //        knative-eventing logging configuration and dynamic updating.  To that end, we are setting up a basic context ourselves
 //        that mirrors what the injection framework would have created.
 //
-func LoggingContext(ctx context.Context, component string, serverUrl string, kubeconfigPath string) context.Context {
-
-	// Get The K8S Client
-	k8sClient := K8sClientWrapper(serverUrl, kubeconfigPath)
-
-	// Put The Kubernetes Client Into The Context Where The Injection Framework Expects It
-	ctx = context.WithValue(ctx, injectionclient.Key{}, k8sClient)
-
+func LoggingContext(ctx context.Context, component string, k8sClient kubernetes.Interface) context.Context {
 	// Get The Logging Config From Knative SharedMain
 	loggingConfig, err := sharedmain.GetLoggingConfig(ctx)
 	if err != nil {
diff --git a/pkg/channel/distributed/receiver/channel/channel.go b/pkg/channel/distributed/receiver/channel/channel.go
index 9a7b43b59..be6113382 100644
--- a/pkg/channel/distributed/receiver/channel/channel.go
+++ b/pkg/channel/distributed/receiver/channel/channel.go
@@ -23,7 +23,6 @@ import (
 
 	"go.uber.org/zap"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
-	k8sclientcmd "k8s.io/client-go/tools/clientcmd"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	kafkainformers "knative.dev/eventing-kafka/pkg/client/informers/externalversions"
@@ -39,33 +38,11 @@ var (
 	stopChan           chan struct{}
 )
 
-// Wrapper Around Kafka Client Creation To Facilitate Unit Testing
-var getKafkaClient = func(ctx context.Context, serverUrl string, kubeconfigPath string) (kafkaclientset.Interface, error) {
-
-	// Create The K8S Configuration (In-Cluster With Cmd Line Flags For Out-Of-Cluster Usage)
-	k8sConfig, err := k8sclientcmd.BuildConfigFromFlags(serverUrl, kubeconfigPath)
-	if err != nil {
-		logging.FromContext(ctx).Error("Failed To Build Kubernetes Config", zap.Error(err))
-		return nil, err
-	}
-
-	// Create A New Kafka Client From The K8S Config & Return The Result
-	return kafkaclientset.NewForConfigOrDie(k8sConfig), nil
-}
-
 // Initialize The KafkaChannel Lister Singleton
-func InitializeKafkaChannelLister(ctx context.Context, serverUrl string, kubeconfigPath string, healthServer *health.Server, resyncDuration time.Duration) error {
-
+func InitializeKafkaChannelLister(ctx context.Context, client kafkaclientset.Interface, healthServer *health.Server, resyncDuration time.Duration) error {
 	// Get The Logger From The Provided Context
 	logger = logging.FromContext(ctx).Desugar()
 
-	// Get The K8S Kafka Client For KafkaChannels
-	client, err := getKafkaClient(ctx, serverUrl, kubeconfigPath)
-	if err != nil {
-		logger.Error("Failed To Create Kafka Client", zap.Error(err))
-		return err
-	}
-
 	// Create A New KafkaChannel SharedInformerFactory For ALL Namespaces (Default Resync Is 10 Hrs)
 	sharedInformerFactory := kafkainformers.NewSharedInformerFactory(client, resyncDuration)
 
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
index 47332745f..1d5ac7203 100644
--- a/pkg/common/scheduler/statefulset/autoscaler.go
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -75,12 +75,14 @@ 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
-				}
-
-				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
+				//}
+				// 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 {
 					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 ffc519265..7122b8c28 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -92,23 +92,22 @@ func (a *Adapter) Start(ctx context.Context) error {
 func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 	a.sourcesMu.Lock()
 	defer a.sourcesMu.Unlock()
+	a.logger.Infow("adding source", "name", obj.Name)
 
 	key := obj.Namespace + "/" + obj.Name
 
-	_, ok := a.sources[key]
+	cancel, ok := a.sources[key]
 
 	if ok {
-		// TODO: see https://github.com/knative-sandbox/eventing-kafka/issues/382
-		return
 		// TODO: do not stop if the only thing that changes is the number of vreplicas
-		//a.logger.Info("stopping adapter", zap.String("key", key))
-		//cancel()
+		a.logger.Info("stopping adapter", zap.String("key", key))
+		cancel()
 	}
 
 	placement := scheduler.GetPlacementForPod(obj.GetPlacements(), a.config.PodName)
 	if placement == nil || placement.VReplicas == 0 {
 		// this pod does not handle this source. Skipping
-		a.logger.Infow("no replicas assigned to this pod. skipping", zap.String("key", key))
+		a.logger.Infow("no replicas assigned to this source. skipping", zap.String("key", key))
 		return
 	}
 
@@ -166,27 +165,30 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 		if err != nil {
 			a.logger.Errorw("adapter failed to start", zap.Error(err))
 		}
-
 	}(ctx)
 
 	a.sources[key] = cancelFn
+	a.logger.Infow("source added", "name", obj.Name)
 }
 
 func (a *Adapter) Remove(ctx context.Context, obj *v1beta1.KafkaSource) {
 	a.sourcesMu.Lock()
 	defer a.sourcesMu.Unlock()
+	a.logger.Infow("removing source", "name", obj.Name)
 
 	key := obj.Namespace + "/" + obj.Name
 
 	cancel, ok := a.sources[key]
 
 	if !ok {
+		a.logger.Infow("source not found", "name", obj.Name)
 		return
 	}
 
 	cancel()
 
 	delete(a.sources, key)
+	a.logger.Infow("source removed", "name", obj.Name, "count", len(a.sources))
 }
 
 // ResolveSecret resolves the secret reference
diff --git a/pkg/source/mtadapter/controller.go b/pkg/source/mtadapter/controller.go
index 7801b9a01..0b13ea335 100644
--- a/pkg/source/mtadapter/controller.go
+++ b/pkg/source/mtadapter/controller.go
@@ -18,6 +18,7 @@ package mtadapter
 
 import (
 	"context"
+	"strings"
 
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
@@ -41,7 +42,7 @@ type MTAdapter interface {
 // NewController initializes the controller and
 // registers event handlers to enqueue events.
 func NewController(ctx context.Context, adapter adapter.Adapter) *controller.Impl {
-	mtadapter, ok := adapter.(MTAdapter)
+	mtadapter, ok := adapter.(*Adapter)
 	if !ok {
 		logging.FromContext(ctx).Fatal("Multi-tenant adapters must implement the MTAdapter interface")
 	}
@@ -50,9 +51,11 @@ 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:],
 		}
 	})
 
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
index bb0127e55..a2be82b20 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
@@ -162,7 +162,7 @@ func RetryConfigFromDeliverySpec(spec duckv1.DeliverySpec) (RetryConfig, error)
 
 // Simple default implementation
 func RetryIfGreaterThan300(_ context.Context, response *nethttp.Response, err error) (bool, error) {
-	return !(response != nil && (response.StatusCode < 300)), err
+	return !(response != nil && (response.StatusCode < 300 && response.StatusCode != -1)), err
 }
 
 // Alternative function to determine whether to retry based on response
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 23f8d0864..de9f273de 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1080,7 +1080,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.20.1-0.20210217232802-fb8e3b49931a
+# knative.dev/eventing v0.20.1-0.20210219125302-4e5bdf97930a
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test


Test patch:
diff --git a/pkg/channel/distributed/common/k8s/context_test.go b/pkg/channel/distributed/common/k8s/context_test.go
index cfd39e8b8..3208203d6 100644
--- a/pkg/channel/distributed/common/k8s/context_test.go
+++ b/pkg/channel/distributed/common/k8s/context_test.go
@@ -21,13 +21,11 @@ import (
 	"testing"
 	"time"
 
-	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
@@ -58,17 +56,10 @@ func TestLoggingContext(t *testing.T) {
 	// Create The Fake K8S Client
 	fakeK8sClient := fake.NewSimpleClientset(loggingConfigMap)
 
-	// Temporarily Swap The K8S Client Wrapper For Testing
-	k8sClientWrapperRef := K8sClientWrapper
-	K8sClientWrapper = func(serverUrlArg string, kubeconfigPathArg string) kubernetes.Interface {
-		assert.Empty(t, serverUrlArg)
-		assert.Empty(t, kubeconfigPathArg)
-		return fakeK8sClient
-	}
-	defer func() { K8sClientWrapper = k8sClientWrapperRef }()
+	ctx = context.WithValue(ctx, kubeclient.Key{}, fakeK8sClient)
 
 	// Perform The Test (Initialize The Logging Context)
-	resultContext := LoggingContext(ctx, component, "", "")
+	resultContext := LoggingContext(ctx, component, fakeK8sClient)
 
 	// Verify The Results
 	assert.NotNil(t, resultContext)
diff --git a/pkg/channel/distributed/receiver/channel/channel_test.go b/pkg/channel/distributed/receiver/channel/channel_test.go
index d6f4d98aa..f8985fd8e 100644
--- a/pkg/channel/distributed/receiver/channel/channel_test.go
+++ b/pkg/channel/distributed/receiver/channel/channel_test.go
@@ -25,7 +25,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	channelhealth "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	receivertesting "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/testing"
-	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	fakeclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
@@ -33,18 +32,14 @@ import (
 
 // Test The InitializeKafkaChannelLister() Functionality
 func TestInitializeKafkaChannelLister(t *testing.T) {
-
-	// Stub The K8S Client Creation Wrapper With Test Version Returning The Fake KafkaClient Clientset
-	getKafkaClient = func(ctx context.Context, serverUrl string, kubeconfigPath string) (kafkaclientset.Interface, error) {
-		return fakeclientset.NewSimpleClientset(), nil
-	}
-
 	// Create A Context With Test Logger
 	ctx := logging.WithLogger(context.TODO(), logtesting.TestLogger(t))
 
+	kafkaClient := fakeclientset.NewSimpleClientset()
+
 	// Perform The Test
 	healthServer := channelhealth.NewChannelHealthServer("12345")
-	err := InitializeKafkaChannelLister(ctx, "", "", healthServer, 600*time.Minute)
+	err := InitializeKafkaChannelLister(ctx, kafkaClient, healthServer, 600*time.Minute)
 
 	// Verify The Results
 	assert.Nil(t, err)
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index afd539e4b..0f0d5719a 100755
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -455,8 +455,21 @@ function test_mt_source() {
   echo "Testing the multi-tenant source"
   install_mt_source || return 1
 
+  export TEST_MT_SOURCE
   go_test_e2e -tags=source,mtsource -timeout=20m -test.parallel=${TEST_PARALLEL} ./test/e2e/...  || fail_test
 
+  # wait for all KafkaSources to be deleted
+  local iterations=0
+  local progress="Waiting for KafkaSources to be deleted..."
+  while [[ "$(kubectl get kafkasources --all-namespaces)" != "" && $iterations -lt 60 ]]
+  do
+    echo -ne "${progress}\r"
+    progress="${progress}."
+    iterations=$((iterations + 1))
+    kubectl get kafkasources --all-namespaces -oyaml
+    sleep 3
+  done
+
   uninstall_mt_source || return 1
 }
 
diff --git a/test/e2e/kafka_source_test.go b/test/e2e/kafka_source_test.go
index d37128244..0bde3da19 100644
--- a/test/e2e/kafka_source_test.go
+++ b/test/e2e/kafka_source_test.go
@@ -23,8 +23,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
 	"strings"
 	"testing"
+	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	. "github.com/cloudevents/sdk-go/v2/test"
@@ -63,6 +65,10 @@ type authSetup struct {
 	TLSEnabled      bool
 }
 
+var (
+	test_mt_source = os.Getenv("TEST_MT_SOURCE")
+)
+
 func withAuthEnablementV1Beta1(auth authSetup) contribresources.KafkaSourceV1Beta1Option {
 	// We test with sasl512 and enable tls with it, so check tls first
 	if auth.TLSEnabled == true {
@@ -244,6 +250,11 @@ func testKafkaSource(t *testing.T, name string, version string, messageKey strin
 
 	client.WaitForAllTestResourcesReadyOrFail(context.Background())
 
+	// See https://github.com/knative-sandbox/eventing-kafka/issues/411
+	if test_mt_source == "1" {
+		time.Sleep(20 * time.Second)
+	}
+
 	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain, kafkaTopicName, messageKey, messageHeaders, messagePayload)
 
 	eventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(cloudEventsSourceName, cloudEventsEventType)))
@@ -534,6 +545,11 @@ func testKafkaSourceUpdate(t *testing.T, name string, test updateTest) {
 	))
 	client.WaitForAllTestResourcesReadyOrFail(context.Background())
 
+	// See https://github.com/knative-sandbox/eventing-kafka/issues/411
+	if test_mt_source == "1" {
+		time.Sleep(20 * time.Second)
+	}
+
 	t.Logf("Send update event to kafkatopic")
 	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain,
 		defaultKafkaSource.topicName+name,
@@ -570,6 +586,11 @@ func testKafkaSourceUpdate(t *testing.T, name string, test updateTest) {
 	contribtestlib.UpdateKafkaSourceV1Beta1OrFail(client, ksObj)
 	client.WaitForAllTestResourcesReadyOrFail(context.Background())
 
+	// See https://github.com/knative-sandbox/eventing-kafka/issues/411
+	if test_mt_source == "1" {
+		time.Sleep(20 * time.Second)
+	}
+
 	t.Logf("Send update event to kafkatopic")
 	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain,
 		test.topicName+name,