Repository: knative-sandbox/eventing-kafka-broker
Pull request #3116 (2023-05-30 14:16:43)

Cherry pick sourcev2 upgrade fix to 1.10
<!-- Please include the 'why' behind your changes if no issue exists -->
## Proposed Changes

- Cherry pick commit https://github.com/knative-sandbox/eventing-kafka-broker/commit/e45c5fec51f1e4af193afdede8efa10d780b0fae to version 1.10

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3116#" title="Author self-approved">Cali0707</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign psschwei for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.10/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":["psschwei"]} -->
Cali0707: /cc @pierDipi 
knative-prow[bot]: @Cali0707: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_release-1.10 | 7331adffae07d342b400ee98c5c77ce635f2b2e1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3116/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_release-1.10/1663550014854533120) | true | `/test channel-reconciler-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<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[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3116#" title="Author self-approved">Cali0707</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3116#pullrequestreview-1451135020" title="Approved">pierDipi</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-broker).

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-broker/blob/release-1.10/OWNERS)~~ [pierDipi]

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/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index fe4aa453b..97441ffd2 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -98,6 +98,11 @@ func GetLabels(name string) map[string]string {
 	}
 }
 
+// Need to have an empty definition here to ensure that we can delete older sources which had a finalizer
+func (r Reconciler) FinalizeKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
+	return nil
+}
+
 func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
 	var deliverySpec *internalscg.DeliverySpec
 	if ks.Spec.Delivery != nil {


Test patch:
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 9d9a5d564..3fe6a9cb9 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -55,6 +55,18 @@ import (
 	kedaclient "knative.dev/eventing-kafka-broker/third_party/pkg/client/injection/client/fake"
 )
 
+const (
+	finalizerName = "kafkasources.sources.knative.dev"
+)
+
+var (
+	finalizerUpdatedEvent = Eventf(
+		corev1.EventTypeNormal,
+		"FinalizerUpdate",
+		fmt.Sprintf(`Updated %q finalizers`, SourceName),
+	)
+)
+
 func TestGetLabels(t *testing.T) {
 
 	testLabels := GetLabels("testSourceName")
@@ -137,6 +149,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal, offset earliest",
@@ -184,6 +202,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal, offset latest",
@@ -231,6 +255,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal with delivery spec",
@@ -283,6 +313,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal, key type label",
@@ -331,6 +367,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal with SASL with type",
@@ -460,6 +502,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal with SASL without type",
@@ -573,6 +621,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - ce overrides",
@@ -628,6 +682,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - with autoscaling annotations",
@@ -676,6 +736,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg with sink update",
@@ -759,6 +825,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg with update",
@@ -818,6 +890,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg with annotations update",
@@ -878,6 +956,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg with update but not ready",
@@ -936,6 +1020,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg without update",
@@ -983,6 +1073,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg without update but not ready",
@@ -1028,6 +1124,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg but failed",
@@ -1074,6 +1176,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - existing cg with replicas set in status",
@@ -1122,6 +1230,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - update cg replicas",
@@ -1204,6 +1318,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 		{
 			Name: "Reconciled normal - ignore source replicas when KEDA is enabled",
@@ -1255,6 +1375,12 @@ func TestReconcileKind(t *testing.T) {
 					),
 				},
 			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
 		},
 	}
 
@@ -1366,3 +1492,12 @@ func SourceNetSaslTls(withType bool) KRShapedOption {
 		}
 	}
 }
+
+func patchFinalizers() clientgotesting.PatchActionImpl {
+	action := clientgotesting.PatchActionImpl{}
+	action.Name = SourceName
+	action.Namespace = SourceNamespace
+	patch := `{"metadata":{"finalizers":["` + finalizerName + `"],"resourceVersion":""}}`
+	action.Patch = []byte(patch)
+	return action
+}