Repository: knative-sandbox/eventing-kafka-broker
Pull request #2352 (2022-06-24 07:36:25)

Adding KafkaSink to storage version migrator
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- adding `kafkasinks.eventing.knative.dev` to `storage-version-migrator` as a NOOP

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: 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**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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/2352#" title="Author self-approved">matzew</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/main/OWNERS)~~ [matzew]

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2352?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2352](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2352?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b264167) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/db9376a6ad6f9783e754ca6ddeac79da24a94798?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (db9376a) will **not change** coverage.
> The diff coverage is `n/a`.

```diff
@@           Coverage Diff           @@
##             main    #2352   +/-   ##
=======================================
  Coverage   59.66%   59.66%           
=======================================
  Files          68       68           
  Lines        6826     6826           
=======================================
  Hits         4073     4073           
  Misses       2467     2467           
  Partials      286      286           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2352?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2352?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [db9376a...b264167](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2352?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow[bot]: @matzew: 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | b264167c73b7c25403724cd2e126a08b550993f2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2352/unit-tests_eventing-kafka-broker_main/1540237403048906752) | true | `/test unit-tests_eventing-kafka-broker_main`

[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]: @matzew: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | b264167c73b7c25403724cd2e126a08b550993f2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2352/unit-tests_eventing-kafka-broker_main/1540237403048906752) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-ssl_eventing-kafka-broker_main | b264167c73b7c25403724cd2e126a08b550993f2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2352/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1540237403367673856) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`

[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]: @matzew: The following tests **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-integration-tests-sasl-ssl_eventing-kafka-broker_main | b264167c73b7c25403724cd2e126a08b550993f2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2352/channel-integration-tests-sasl-ssl_eventing-kafka-broker_main/1540237403367673856) | true | `/test channel-integration-tests-sasl-ssl_eventing-kafka-broker_main`
unit-tests_eventing-kafka-broker_main | b264167c73b7c25403724cd2e126a08b550993f2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2352/unit-tests_eventing-kafka-broker_main/1540267404389519360) | true | `/test unit-tests_eventing-kafka-broker_main`

[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 -->

Patch:
diff --git a/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml b/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml
index cec9e3b19..1c4283b58 100644
--- a/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml
+++ b/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml
@@ -58,6 +58,19 @@ rules:
       - "update"
       - "patch"
       - "watch"
+  - apiGroups:
+      - "eventing.knative.dev"
+    resources:
+      - "kafkasinks"
+      - "kafkasinks/finalizers"
+      - "kafkasinks/status"
+    verbs:
+      - "get"
+      - "list"
+      - "create"
+      - "update"
+      - "patch"
+      - "watch"
   - apiGroups:
       - ""
     resources:
diff --git a/control-plane/config/post-install/500-storage-version-migrator.yaml b/control-plane/config/post-install/500-storage-version-migrator.yaml
index a7003b860..26a3b70e0 100644
--- a/control-plane/config/post-install/500-storage-version-migrator.yaml
+++ b/control-plane/config/post-install/500-storage-version-migrator.yaml
@@ -42,3 +42,4 @@ spec:
           args:
             - "kafkasources.sources.knative.dev"
             - "kafkachannels.messaging.knative.dev"
+            - "kafkasinks.eventing.knative.dev"
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 516eee779..0b4e738c8 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -81,7 +81,7 @@ type Reconciler struct {
 	DataPlaneConfigMapNamespace string
 	DataPlaneConfigMapName      string
 	DataPlaneConfigFormat       string
-	SystemNamespace             string
+	DataPlaneNamespace          string
 
 	DispatcherLabel string
 	ReceiverLabel   string
@@ -222,17 +222,17 @@ func (r *Reconciler) UpdateDataPlaneConfigMap(ctx context.Context, contract *con
 }
 
 func (r *Reconciler) UpdateDispatcherPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
-	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.dispatcherSelector())
+	pods, errors := r.PodLister.Pods(r.DataPlaneNamespace).List(r.dispatcherSelector())
 	if errors != nil {
-		return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.SystemNamespace, errors)
+		return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.DataPlaneNamespace, errors)
 	}
 	return r.UpdatePodsAnnotation(ctx, logger, "dispatcher", volumeGeneration, pods)
 }
 
 func (r *Reconciler) UpdateReceiverPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
-	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.ReceiverSelector())
+	pods, errors := r.PodLister.Pods(r.DataPlaneNamespace).List(r.ReceiverSelector())
 	if errors != nil {
-		return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
+		return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.DataPlaneNamespace, errors)
 	}
 	return r.UpdatePodsAnnotation(ctx, logger, "receiver", volumeGeneration, pods)
 }
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 3074b2747..4a9931de0 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -65,7 +65,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 			DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
 			DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
 			DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
-			SystemNamespace:             env.SystemNamespace,
+			DataPlaneNamespace:          env.SystemNamespace,
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
 		},
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 95ce83b6c..2cab70d39 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -38,12 +38,10 @@ import (
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
 
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
-	"knative.dev/pkg/system"
-
-	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
@@ -545,7 +543,7 @@ func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messaging
 }
 
 func (r *Reconciler) channelConfigMap() (*corev1.ConfigMap, error) {
-	namespace := system.Namespace()
+	namespace := r.DataPlaneNamespace
 	cm, err := r.ConfigMapLister.ConfigMaps(namespace).Get(r.Env.GeneralConfigMapName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, r.Env.GeneralConfigMapName, err)
@@ -612,7 +610,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 }
 
 func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) (*corev1.Service, error) {
-	expected, err := resources.MakeK8sService(channel, resources.ExternalService(system.Namespace(), NewChannelIngressServiceName))
+	expected, err := resources.MakeK8sService(channel, resources.ExternalService(r.DataPlaneNamespace, NewChannelIngressServiceName))
 	if err != nil {
 		return expected, fmt.Errorf("failed to create the channel service object: %w", err)
 	}
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index c45d597a1..9dcf79f7f 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -61,7 +61,7 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
 			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
 			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-			SystemNamespace:             configs.SystemNamespace,
+			DataPlaneNamespace:          configs.SystemNamespace,
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2.go b/control-plane/pkg/reconciler/channel/v2/channelv2.go
index b07024c9f..7f8241283 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2.go
@@ -39,16 +39,14 @@ import (
 	"knative.dev/pkg/network"
 	"knative.dev/pkg/resolver"
 
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/reconciler"
-	"knative.dev/pkg/system"
-
-	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -309,7 +307,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta
 }
 
 func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) (*corev1.Service, error) {
-	expected, err := resources.MakeK8sService(channel, resources.ExternalService(r.Env.SystemNamespace, channelreconciler.NewChannelIngressServiceName))
+	expected, err := resources.MakeK8sService(channel, resources.ExternalService(r.DataPlaneNamespace, channelreconciler.NewChannelIngressServiceName))
 	if err != nil {
 		return expected, fmt.Errorf("failed to create the channel service object: %w", err)
 	}
@@ -654,7 +652,7 @@ func consumerGroup(channel *messagingv1beta1.KafkaChannel, s *v1.SubscriberSpec)
 func (r *Reconciler) channelConfigMap() (*corev1.ConfigMap, error) {
 	// TODO: do we want to support namespaced channels? they're not supported at the moment.
 
-	namespace := system.Namespace()
+	namespace := r.DataPlaneNamespace
 	cm, err := r.ConfigMapLister.ConfigMaps(namespace).Get(r.Env.GeneralConfigMapName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, r.Env.GeneralConfigMapName, err)
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2.go b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
index 671dffa3c..db8befc7c 100644
--- a/control-plane/pkg/reconciler/channel/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
@@ -63,7 +63,7 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
 			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
 			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-			SystemNamespace:             configs.SystemNamespace,
+			DataPlaneNamespace:          configs.SystemNamespace,
 		},
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index dd708d1ec..754a93057 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -379,7 +379,7 @@ func (r Reconciler) commonReconciler(p *corev1.Pod, cmName string) base.Reconcil
 		DataPlaneConfigMapNamespace: p.GetNamespace(),
 		DataPlaneConfigMapName:      cmName,
 		DataPlaneConfigFormat:       string(r.SerDe.Format),
-		SystemNamespace:             p.GetNamespace(),
+		DataPlaneNamespace:          p.GetNamespace(),
 		DispatcherLabel:             "",
 		ReceiverLabel:               "",
 	}
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index b0de02839..bf7f086ab 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -59,7 +59,7 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
 			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
 			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-			SystemNamespace:             configs.SystemNamespace,
+			DataPlaneNamespace:          configs.SystemNamespace,
 			ReceiverLabel:               base.SinkReceiverLabel,
 		},
 		ConfigMapLister:            configmapInformer.Lister(),
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index 851905f84..824fd6c28 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -68,7 +68,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
 			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
 			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-			SystemNamespace:             configs.SystemNamespace,
+			DataPlaneNamespace:          configs.SystemNamespace,
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
 		},
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index a44012b0c..5ed225d47 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -42,8 +42,8 @@ Lists of 158 third-party dependencies.
      (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
      (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
      (The Apache Software License, Version 2.0) cloudevents-sql (io.cloudevents:cloudevents-sql:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-sql/)
-     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.9.1.Final - https://debezium.io/debezium-parent/debezium-api)
-     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.9.1.Final - https://debezium.io/debezium-parent/debezium-core)
+     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.9.4.Final - https://debezium.io/debezium-parent/debezium-api)
+     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.9.4.Final - https://debezium.io/debezium-parent/debezium-core)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.12.2 - http://fabric8.io/kubernetes-client/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.12.2 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.12.2 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 69f8ba0db..b70b8dfd2 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -57,7 +57,7 @@
     <mokito.junit.jupiter.version>4.6.0</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.12.2</fabric8.kubernetes.version>
     <kafka.version>2.8.1</kafka.version>
-    <debezium.version>1.9.1.Final</debezium.version>
+    <debezium.version>1.9.4.Final</debezium.version>
     <jib.version>3.2.0</jib.version>
   </properties>
 


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 4b53865b4..d2622c32c 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -2217,7 +2217,7 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
 				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
 				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
-				SystemNamespace:             env.SystemNamespace,
+				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
 			},
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 1414e6dbc..2745e2bef 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -1623,7 +1623,7 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
 				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
 				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
-				SystemNamespace:             env.SystemNamespace,
+				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
index 8c738f201..c872e4ba6 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
@@ -29,14 +29,12 @@ import (
 	"k8s.io/utils/pointer"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/network"
 	. "knative.dev/pkg/reconciler/testing"
-	"knative.dev/pkg/system"
-
-	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
@@ -110,7 +108,7 @@ func TestReconcileKind(t *testing.T) {
 				NewChannel(
 					WithInitKafkaChannelConditions,
 					WithDeletedTimeStamp),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -126,7 +124,7 @@ func TestReconcileKind(t *testing.T) {
 				NewChannel(
 					WithInitKafkaChannelConditions,
 					WithDeletedTimeStamp),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -143,7 +141,7 @@ func TestReconcileKind(t *testing.T) {
 				NewChannel(
 					WithInitKafkaChannelConditions,
 					WithDeletedTimeStamp),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapFromContract(&contract.Contract{
@@ -174,7 +172,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled normal - no subscription (no cons group) - no auth",
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
@@ -245,7 +243,7 @@ func TestReconcileKind(t *testing.T) {
 						Retry:          pointer.Int32(5),
 					}),
 				),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
@@ -320,7 +318,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled failed - no subscription (no cons group) - probe " + prober.StatusNotReady.String(),
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 			},
@@ -372,7 +370,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled failed - no subscription (no cons group) - probe " + prober.StatusUnknown.String(),
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 			},
@@ -424,7 +422,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled normal - with single fresh subscriber - cg unknown - no auth",
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -491,7 +489,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled normal - with single unready subscriber - cg unknown - no auth",
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithUnreadySubscriber))),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -558,7 +556,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled normal - with single ready subscriber - cg unknown - no auth",
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1())),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -625,7 +623,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled normal - with single fresh subscriber - existing ready cg with update - no auth",
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -701,7 +699,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled normal - with single fresh subscriber - existing cg but failed - no auth",
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -771,7 +769,7 @@ func TestReconcileKind(t *testing.T) {
 					WithSubscribers(Subscriber1(WithFreshSubscriber)),
 					WithSubscribers(Subscriber2(WithFreshSubscriber)),
 				),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -856,7 +854,7 @@ func TestReconcileKind(t *testing.T) {
 				NewChannel(
 					WithSubscribers(Subscriber1(WithFreshSubscriber)),
 				),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -960,7 +958,7 @@ func TestReconcileKind(t *testing.T) {
 				{
 					Object: NewChannel(
 						WithInitKafkaChannelConditions,
-						StatusConfigNotParsed(fmt.Sprintf(`failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), DefaultEnv.GeneralConfigMapName, DefaultEnv.GeneralConfigMapName)),
+						StatusConfigNotParsed(fmt.Sprintf(`failed to get configmap %s/%s: configmap %q not found`, env.SystemNamespace, DefaultEnv.GeneralConfigMapName, DefaultEnv.GeneralConfigMapName)),
 					),
 				},
 			},
@@ -972,7 +970,7 @@ func TestReconcileKind(t *testing.T) {
 				Eventf(
 					corev1.EventTypeWarning,
 					"InternalError",
-					fmt.Sprintf(`failed to get contract configuration: failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), DefaultEnv.GeneralConfigMapName, DefaultEnv.GeneralConfigMapName),
+					fmt.Sprintf(`failed to get contract configuration: failed to get configmap %s/%s: configmap %q not found`, env.SystemNamespace, DefaultEnv.GeneralConfigMapName, DefaultEnv.GeneralConfigMapName),
 				),
 			},
 		},
@@ -980,7 +978,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "channel configmap does not have bootstrap servers",
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					"foo": "bar",
 				}),
 			},
@@ -1010,7 +1008,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "channel configmap has blank bootstrap servers",
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: "",
 				}),
 			},
@@ -1045,7 +1043,7 @@ func TestReconcileKind(t *testing.T) {
 					WithReplicationFactor(4),
 					WithRetentionDuration("1000"),
 				),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 			},
@@ -1102,7 +1100,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Reconciled normal - with single fresh subscriber - with auth - PlainText",
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 					security.AuthSecretNameKey:         "secret-1",
 					security.AuthSecretNamespaceKey:    "ns-1",
@@ -1181,7 +1179,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Create contract configmap when it does not exist",
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 			},
@@ -1231,7 +1229,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Do not create contract configmap when it exists",
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
@@ -1280,7 +1278,7 @@ func TestReconcileKind(t *testing.T) {
 			Name: "Corrupt contract in configmap",
 			Objects: []runtime.Object{
 				NewChannel(),
-				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, []byte("corrupt")),
@@ -1340,7 +1338,7 @@ func TestReconcileKind(t *testing.T) {
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
 				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
 				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
-				SystemNamespace:             env.SystemNamespace,
+				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 7ad5464ea..6dd65342c 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -1325,7 +1325,7 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
 				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
 				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
-				SystemNamespace:             env.SystemNamespace,
+				DataPlaneNamespace:          env.SystemNamespace,
 				ReceiverLabel:               base.SinkReceiverLabel,
 			},
 			ConfigMapLister: listers.GetConfigMapLister(),
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 66a1cbdff..0adea482e 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -2760,7 +2760,7 @@ func useTableWithFlags(t *testing.T, table TableTest, env *config.Env, flags fea
 				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
 				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
 				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
-				SystemNamespace:             env.SystemNamespace,
+				DataPlaneNamespace:          env.SystemNamespace,
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
 			},