Repository: knative-sandbox/eventing-kafka
Pull request #416 (2021-02-25 14:53:50)

[mtsource] remove finalizer in mtadapter
Similar to https://github.com/knative/eventing/pull/4511

Fixes https://github.com/knative-sandbox/eventing-kafka/issues/412

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- The mt source adapter doesn't set any finalizers anymore, since updating the internal state never fails.
- Reenable scaling down
-

<!--
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
- 🧽 The mt source adapter does not set the finalizers anymore
```

**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 **lionelvillard** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @lionelvillard` in a comment when ready.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["lionelvillard"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416?src=pr&el=h1) Report
> Merging [#416](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416?src=pr&el=desc) (7768c5a) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/c7ab439c66df5641d312c59731a8d33ac0bb8b1a?el=desc) (c7ab439) will **decrease** coverage by `0.28%`.
> The diff coverage is `0.00%`.

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

```diff
@@            Coverage Diff             @@
##           master     #416      +/-   ##
==========================================
- Coverage   73.13%   72.84%   -0.29%     
==========================================
  Files         129      129              
  Lines        4973     4983      +10     
==========================================
- Hits         3637     3630       -7     
- Misses       1100     1117      +17     
  Partials      236      236              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `70.11% <ø> (ø)` | |
| [pkg/source/mtadapter/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvY29udHJvbGxlci5nbw==) | `0.00% <0.00%> (ø)` | |
| [pkg/source/mtadapter/kafkasource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIva2Fma2Fzb3VyY2UuZ28=) | `0.00% <0.00%> (ø)` | |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416/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/416?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/416?src=pr&el=footer). Last update [c7ab439...7768c5a](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/416?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

knative-prow-robot: @lionelvillard: 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 | 73db76581003eb982f3ca3d5803ea110f54baea0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/416/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1364953390253084672) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<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/common/scheduler/statefulset/autoscaler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/416/pull-knative-sandbox-eventing-kafka-go-coverage/1364970063924301824/artifacts/line-cov.html#file0) | 52.5% | 50.0% | -2.5

steven0711dong: /lgtm
lionelvillard: /approve
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/416#issuecomment-786092325" title="Approved">lionelvillard</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/416#issuecomment-786092325" title="Approved">lionelvillard</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/416#pullrequestreview-598854062" title="Approved">steven0711dong</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->

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