Repository: knative-sandbox/eventing-kafka-broker
Pull request #1916 (2022-02-16 11:55:20)

[release-1.2] Add expected status to prober
This is an automated cherry-pick of #1914

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1916#" title="Author self-approved">knative-prow-robot</a>*
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-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.2/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["aliok"]} -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1916#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1916#pullrequestreview-884424573" title="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/release-1.2/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":[]} -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1916#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1916#pullrequestreview-884424573" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1916#pullrequestreview-884425443" 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.2/OWNERS)~~ [matzew,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":[]} -->
knative-prow-robot: @knative-prow-robot: 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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 98de90087a74232b356ec1437c5f6d382f12aaec | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1916/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493916907500212224) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1916). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-prow-robot).

<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-broker/pull/1916?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-1.2@44e4811`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1916/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1916?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@              Coverage Diff               @@
##             release-1.2    #1916   +/-   ##
==============================================
  Coverage               ?   82.01%           
  Complexity             ?      614           
==============================================
  Files                  ?       69           
  Lines                  ?     2024           
  Branches               ?      178           
==============================================
  Hits                   ?     1660           
  Misses                 ?      270           
  Partials               ?       94           
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.01% <0.00%> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.


------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1916?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/1916?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [44e4811...98de900](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1916?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-robot: @knative-prow-robot: 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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 98de90087a74232b356ec1437c5f6d382f12aaec | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1916/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1493916907500212224) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 98de90087a74232b356ec1437c5f6d382f12aaec | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1916/pull-knative-sandbox-eventing-kafka-broker-build-tests/1493916907462463488) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1916). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-prow-robot).

<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: @knative-prow-robot: 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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | 98de90087a74232b356ec1437c5f6d382f12aaec | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1916/pull-knative-sandbox-eventing-kafka-broker-build-tests/1493916907462463488) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl | 98de90087a74232b356ec1437c5f6d382f12aaec | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1916/pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl/1493916907810590720) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-reconciler-tests-ssl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1916). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-prow-robot).

<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 -->
pierDipi: /retest-required

Patch:
diff --git a/.gitignore b/.gitignore
index ccbc1fcce..c81faa7b7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -34,6 +34,8 @@ eventing-kafka-broker.yaml
 eventing-kafka-channel.yaml
 eventing-kafka-sink.yaml
 
+eventing-kafka-post-install.yaml
+
 eventing-kafka-controller-prometheus-operator.yaml
 eventing-kafka-source-prometheus-operator.yaml
 eventing-kafka-channel-prometheus-operator.yaml
diff --git a/control-plane/cmd/post-install/kafka_source_deleter.go b/control-plane/cmd/post-install/kafka_source_deleter.go
new file mode 100644
index 000000000..0ad5f1672
--- /dev/null
+++ b/control-plane/cmd/post-install/kafka_source_deleter.go
@@ -0,0 +1,140 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"fmt"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"knative.dev/pkg/system"
+)
+
+type kafkaSourceDeleter struct {
+	k8s kubernetes.Interface
+}
+
+func (d *kafkaSourceDeleter) Delete(ctx context.Context) error {
+	// Delete mutatingwebhookconfigurations.admissionregistration.k8s.io/defaulting.webhook.kafka.sources.knative.dev
+	const sourceMutatingWebhookConfigurationName = "defaulting.webhook.kafka.sources.knative.dev"
+	err := d.k8s.
+		AdmissionregistrationV1().
+		MutatingWebhookConfigurations().
+		Delete(ctx, sourceMutatingWebhookConfigurationName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete mutating webhook configuration %s: %w", sourceMutatingWebhookConfigurationName, err)
+	}
+
+	// Delete validatingwebhookconfigurations.admissionregistration.k8s.io/validation.webhook.kafka.sources.knative.dev
+	const sourceValidatingWebhookConfigurationName = "validation.webhook.kafka.sources.knative.dev"
+	err = d.k8s.
+		AdmissionregistrationV1().
+		ValidatingWebhookConfigurations().
+		Delete(ctx, sourceValidatingWebhookConfigurationName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete validating webhook configuration %s: %w", sourceMutatingWebhookConfigurationName, err)
+	}
+
+	// Delete validatingwebhookconfigurations.admissionregistration.k8s.io/config.webhook.kafka.sources.knative.dev
+	const sourceConfigValidatingWebhookConfigurationName = "config.webhook.kafka.sources.knative.dev"
+	err = d.k8s.
+		AdmissionregistrationV1().
+		ValidatingWebhookConfigurations().
+		Delete(ctx, sourceConfigValidatingWebhookConfigurationName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete validating webhook configuration %s: %w", sourceConfigValidatingWebhookConfigurationName, err)
+	}
+
+	// Delete service knative-eventing/kafka-source-webhook
+	const sourceWebhookServiceName = "kafka-source-webhook"
+	err = d.k8s.
+		CoreV1().
+		Services(system.Namespace()).
+		Delete(ctx, sourceWebhookServiceName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service %s/%s: %w", system.Namespace(), sourceWebhookServiceName, err)
+	}
+
+	// Delete service knative-eventing/kafka-controller
+	const sourceControllerServiceName = "kafka-controller"
+	err = d.k8s.
+		CoreV1().
+		Services(system.Namespace()).
+		Delete(ctx, sourceControllerServiceName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service %s/%s: %w", system.Namespace(), sourceControllerServiceName, err)
+	}
+
+	// Delete deployment.apps/kafka-controller-manager
+	const sourceControllerDeploymentName = "kafka-controller-manager"
+	err = d.k8s.
+		AppsV1().
+		Deployments(system.Namespace()).
+		Delete(ctx, sourceControllerDeploymentName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete deployment %s/%s: %w", system.Namespace(), sourceControllerDeploymentName, err)
+	}
+
+	const kafkaControllerClusterRoleBinding = "eventing-sources-kafka-controller"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoleBindings().
+		Delete(ctx, kafkaControllerClusterRoleBinding, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRoleBinding %s: %w", kafkaControllerClusterRoleBinding, err)
+	}
+
+	const kafkaControllerAddressableResolverClusterRoleBinding = "eventing-sources-kafka-controller-addressable-resolver"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoleBindings().
+		Delete(ctx, kafkaControllerAddressableResolverClusterRoleBinding, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRoleBinding %s: %w", kafkaControllerAddressableResolverClusterRoleBinding, err)
+	}
+
+	const kafkaControllerClusterRole = "eventing-sources-kafka-controller"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoles().
+		Delete(ctx, kafkaControllerClusterRole, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRole %s: %w", kafkaControllerClusterRole, err)
+	}
+
+	const kafkaControllerServiceAccount = "kafka-controller-manager"
+	err = d.k8s.
+		CoreV1().
+		ServiceAccounts(system.Namespace()).
+		Delete(ctx, kafkaControllerServiceAccount, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service account %s/%s: %w", system.Namespace(), kafkaControllerServiceAccount, err)
+	}
+
+	const kafkaSourceWebhookSecret = "kafka-source-webhook-certs"
+	err = d.k8s.
+		CoreV1().
+		Secrets(system.Namespace()).
+		Delete(ctx, kafkaSourceWebhookSecret, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete secret %s/%s: %w", system.Namespace(), kafkaSourceWebhookSecret, err)
+	}
+
+	return nil
+}
diff --git a/control-plane/cmd/post-install/kafka_source_migrator.go b/control-plane/cmd/post-install/kafka_source_migrator.go
new file mode 100644
index 000000000..85e5af09c
--- /dev/null
+++ b/control-plane/cmd/post-install/kafka_source_migrator.go
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
+	kcs "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+)
+
+type KafkaSourceMigrator struct {
+	kcs kcs.Interface
+	k8s kubernetes.Interface
+}
+
+func (m *KafkaSourceMigrator) Migrate(ctx context.Context) error {
+
+	// cont takes care of results pagination.
+	cont := ""
+	first := true
+	for first || cont != "" {
+		kss, err := m.kcs.
+			SourcesV1beta1().
+			KafkaSources(corev1.NamespaceAll).
+			List(ctx, metav1.ListOptions{Continue: cont})
+		if err != nil {
+			return fmt.Errorf("failed to list KafkaSources: %w", err)
+		}
+
+		for _, ks := range kss.Items {
+
+			// The KafkaSource deployment name isn't predictable, so we use a label selector to get
+			// the associated deployment.
+			ksLabels := map[string]string{
+				// See https://github.com/knative-sandbox/eventing-kafka/blob/2b185a9b59347cc4a16471ed0fb2ba3bdf6e1672/pkg/source/reconciler/source/resources/labels.go#L24-L39
+				"eventing.knative.dev/source":     "kafka-source-controller",
+				"eventing.knative.dev/sourceName": ks.GetName(),
+			}
+			lsOpts := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(ksLabels).String()}
+
+			// Instead of listing and then deleting a single deployment,
+			// we use delete collection that allows deleting objects based
+			// on a label selector.
+			pp := metav1.DeletePropagationForeground
+			err := m.k8s.AppsV1().
+				Deployments(ks.GetNamespace()).
+				DeleteCollection(ctx, metav1.DeleteOptions{PropagationPolicy: &pp}, lsOpts)
+			if err != nil && !apierrors.IsNotFound(err) {
+				return fmt.Errorf("failed to delete deployments with selector %v: %w", lsOpts.LabelSelector, err)
+			}
+		}
+
+		cont = kss.Continue
+		first = false
+	}
+
+	return nil
+}
diff --git a/control-plane/cmd/post-install/main.go b/control-plane/cmd/post-install/main.go
new file mode 100644
index 000000000..90e3cdd36
--- /dev/null
+++ b/control-plane/cmd/post-install/main.go
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+
+	"k8s.io/client-go/kubernetes"
+	kcs "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	"knative.dev/pkg/environment"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/signals"
+)
+
+func main() {
+	ctx := signals.NewContext()
+
+	config, err := logging.NewConfigFromMap(nil)
+	if err != nil {
+		log.Fatal("Failed to create logging config: ", err)
+	}
+
+	logger, _ := logging.NewLoggerFromConfig(config, "kafka-broker-post-install")
+	defer logger.Sync()
+
+	logging.WithLogger(ctx, logger)
+
+	if err := run(ctx); err != nil {
+		logger.Fatal(err)
+	}
+}
+
+func run(ctx context.Context) error {
+
+	env := environment.ClientConfig{}
+	env.InitFlags(flag.CommandLine)
+	flag.Parse()
+
+	config, err := env.GetRESTConfig()
+	if err != nil {
+		return fmt.Errorf("failed to get kubeconfig: %w", err)
+	}
+
+	sourceMigrator := &KafkaSourceMigrator{
+		kcs: kcs.NewForConfigOrDie(config),
+		k8s: kubernetes.NewForConfigOrDie(config),
+	}
+	if err := sourceMigrator.Migrate(ctx); err != nil {
+		return fmt.Errorf("source migration failed: %w", err)
+	}
+
+	sourceDeleter := &kafkaSourceDeleter{
+		k8s: kubernetes.NewForConfigOrDie(config),
+	}
+	if err := sourceDeleter.Delete(ctx); err != nil {
+		return fmt.Errorf("source migration failed: %w", err)
+	}
+
+	return nil
+}
diff --git a/control-plane/config/100-channel/100-kafka-channel.yaml b/control-plane/config/100-channel/100-kafka-channel.yaml
deleted file mode 120000
index d60cf59f1..000000000
--- a/control-plane/config/100-channel/100-kafka-channel.yaml
+++ /dev/null
@@ -1,1 +0,0 @@
-../../../third_party/eventing-kafka-latest/channel-crds.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-source/100-kafka-source.yaml b/control-plane/config/100-source/100-kafka-source.yaml
deleted file mode 120000
index 1e45756e5..000000000
--- a/control-plane/config/100-source/100-kafka-source.yaml
+++ /dev/null
@@ -1,1 +0,0 @@
-../../../third_party/eventing-kafka-latest/source-crd.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-broker/100-kafka-broker-configmap.yaml b/control-plane/config/eventing-kafka-broker/100-broker/100-kafka-broker-configmap.yaml
similarity index 100%
rename from control-plane/config/100-broker/100-kafka-broker-configmap.yaml
rename to control-plane/config/eventing-kafka-broker/100-broker/100-kafka-broker-configmap.yaml
diff --git a/control-plane/config/100-channel/100-kafka-channel-configmap.yaml b/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel-configmap.yaml
similarity index 100%
rename from control-plane/config/100-channel/100-kafka-channel-configmap.yaml
rename to control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel-configmap.yaml
diff --git a/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml b/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml
new file mode 120000
index 000000000..0f2cef1b2
--- /dev/null
+++ b/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml
@@ -0,0 +1,1 @@
+../../../../third_party/eventing-kafka-latest/channel-crds.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-kafka-internal/100-consumer.yaml b/control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumer.yaml
similarity index 100%
rename from control-plane/config/100-kafka-internal/100-consumer.yaml
rename to control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumer.yaml
diff --git a/control-plane/config/100-kafka-internal/100-consumergroup.yaml b/control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumergroup.yaml
similarity index 100%
rename from control-plane/config/100-kafka-internal/100-consumergroup.yaml
rename to control-plane/config/eventing-kafka-broker/100-kafka-internal/100-consumergroup.yaml
diff --git a/control-plane/config/100-sink/100-kafka-sink.yaml b/control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml
similarity index 100%
rename from control-plane/config/100-sink/100-kafka-sink.yaml
rename to control-plane/config/eventing-kafka-broker/100-sink/100-kafka-sink.yaml
diff --git a/control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml b/control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml
new file mode 120000
index 000000000..595cfd702
--- /dev/null
+++ b/control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml
@@ -0,0 +1,1 @@
+../../../../third_party/eventing-kafka-latest/source-crd.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-source/200-sources-observer-aggregated-clusterrole.yaml b/control-plane/config/eventing-kafka-broker/100-source/200-sources-observer-aggregated-clusterrole.yaml
similarity index 100%
rename from control-plane/config/100-source/200-sources-observer-aggregated-clusterrole.yaml
rename to control-plane/config/eventing-kafka-broker/100-source/200-sources-observer-aggregated-clusterrole.yaml
diff --git a/control-plane/config/200-controller/100-config-kafka-leader-election.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-kafka-leader-election.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
diff --git a/control-plane/config/200-controller/100-config-logging.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-logging.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml
diff --git a/control-plane/config/200-controller/100-config-tracing.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml
similarity index 100%
rename from control-plane/config/200-controller/100-config-tracing.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml
diff --git a/control-plane/config/200-controller/200-addressable-resolver-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-addressable-resolver-cluster-role.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-addressable-resolver-cluster-role.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-addressable-resolver-cluster-role.yaml
diff --git a/control-plane/config/200-controller/200-channelable-manipulator-clusterrole.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-channelable-manipulator-clusterrole.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-channelable-manipulator-clusterrole.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-channelable-manipulator-clusterrole.yaml
diff --git a/control-plane/config/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-controller-cluster-role.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
diff --git a/control-plane/config/200-controller/200-controller-service-account.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-service-account.yaml
similarity index 100%
rename from control-plane/config/200-controller/200-controller-service-account.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/200-controller-service-account.yaml
diff --git a/control-plane/config/200-controller/201-controller-cluster-role-binding.yaml b/control-plane/config/eventing-kafka-broker/200-controller/201-controller-cluster-role-binding.yaml
similarity index 100%
rename from control-plane/config/200-controller/201-controller-cluster-role-binding.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/201-controller-cluster-role-binding.yaml
diff --git a/control-plane/config/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
similarity index 100%
rename from control-plane/config/200-controller/500-controller.yaml
rename to control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
diff --git a/control-plane/config/200-webhook/100-webhook-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-cluster-role.yaml
similarity index 100%
rename from control-plane/config/200-webhook/100-webhook-cluster-role.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-cluster-role.yaml
diff --git a/control-plane/config/200-webhook/100-webhook-service-account.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-service-account.yaml
similarity index 100%
rename from control-plane/config/200-webhook/100-webhook-service-account.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-service-account.yaml
diff --git a/control-plane/config/200-webhook/200-webhook-cluster-role-binding.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/200-webhook-cluster-role-binding.yaml
similarity index 100%
rename from control-plane/config/200-webhook/200-webhook-cluster-role-binding.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/200-webhook-cluster-role-binding.yaml
diff --git a/control-plane/config/200-webhook/400-webhook-defaulting.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-defaulting.yaml
similarity index 100%
rename from control-plane/config/200-webhook/400-webhook-defaulting.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-defaulting.yaml
diff --git a/control-plane/config/200-webhook/400-webhook-secret.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-secret.yaml
similarity index 100%
rename from control-plane/config/200-webhook/400-webhook-secret.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-secret.yaml
diff --git a/control-plane/config/200-webhook/400-webhook-validation.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-validation.yaml
similarity index 100%
rename from control-plane/config/200-webhook/400-webhook-validation.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-validation.yaml
diff --git a/control-plane/config/200-webhook/500-webhook.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/500-webhook.yaml
similarity index 100%
rename from control-plane/config/200-webhook/500-webhook.yaml
rename to control-plane/config/eventing-kafka-broker/200-webhook/500-webhook.yaml
diff --git a/control-plane/config/post-install/200-controller-cluster-role.yaml b/control-plane/config/post-install/200-controller-cluster-role.yaml
new file mode 100644
index 000000000..9b1d9ff2b
--- /dev/null
+++ b/control-plane/config/post-install/200-controller-cluster-role.yaml
@@ -0,0 +1,114 @@
+---
+
+# Copyright 2022 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: knative-kafka-controller-post-install
+  labels:
+    kafka.eventing.knative.dev/release: devel
+rules:
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      - delete
+      - deletecollection
+
+  # sources.knative.dev resources and finalizers we care about.
+  - apiGroups:
+      - sources.knative.dev
+    resources:
+      - kafkasources
+      - kafkasources/status
+    verbs:
+      - get
+      - list
+
+  - apiGroups:
+      - admissionregistration.k8s.io
+    resources:
+      - mutatingwebhookconfigurations
+    resourceNames:
+      - defaulting.webhook.kafka.sources.knative.dev
+    verbs:
+      - delete
+
+  - apiGroups:
+      - admissionregistration.k8s.io
+    resources:
+      - validatingwebhookconfigurations
+    resourceNames:
+      - validation.webhook.kafka.sources.knative.dev
+      - config.webhook.kafka.sources.knative.dev
+    verbs:
+      - delete
+
+  - apiGroups:
+      - "*"
+    resources:
+      - services
+    resourceNames:
+      - kafka-source-webhook
+      - kafka-controller
+    verbs:
+      - delete
+
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    resourceNames:
+      - kafka-controller-manager
+    verbs:
+      - delete
+
+  - apiGroups:
+      - rbac.authorization.k8s.io
+    resources:
+      - clusterrolebindings
+    resourceNames:
+      - eventing-sources-kafka-controller
+      - eventing-sources-kafka-controller-addressable-resolver
+    verbs:
+      - delete
+  - apiGroups:
+      - rbac.authorization.k8s.io
+    resources:
+      - clusterroles
+    resourceNames:
+      - eventing-sources-kafka-controller
+    verbs:
+      - delete
+
+  - apiGroups:
+      - "*"
+    resources:
+      - serviceaccounts
+    resourceNames:
+      - kafka-controller-manager
+    verbs:
+      - delete
+
+  - apiGroups:
+      - "*"
+    resources:
+      - secrets
+    resourceNames:
+      - kafka-source-webhook-certs
+    verbs:
+      - delete
diff --git a/control-plane/config/post-install/200-controller-service-account.yaml b/control-plane/config/post-install/200-controller-service-account.yaml
new file mode 100644
index 000000000..91f65a3d9
--- /dev/null
+++ b/control-plane/config/post-install/200-controller-service-account.yaml
@@ -0,0 +1,23 @@
+---
+
+# Copyright 2022 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: knative-kafka-controller-post-install
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
diff --git a/control-plane/config/post-install/201-controller-cluster-role-binding.yaml b/control-plane/config/post-install/201-controller-cluster-role-binding.yaml
new file mode 100644
index 000000000..9f36b0c8c
--- /dev/null
+++ b/control-plane/config/post-install/201-controller-cluster-role-binding.yaml
@@ -0,0 +1,30 @@
+---
+
+# Copyright 2022 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: knative-kafka-controller-post-install
+  labels:
+    kafka.eventing.knative.dev/release: devel
+subjects:
+  - kind: ServiceAccount
+    name: knative-kafka-controller-post-install
+    namespace: knative-eventing
+roleRef:
+  kind: ClusterRole
+  name: knative-kafka-controller-post-install
+  apiGroup: rbac.authorization.k8s.io
diff --git a/control-plane/config/post-install/500-post-install-job.yaml b/control-plane/config/post-install/500-post-install-job.yaml
new file mode 100644
index 000000000..53f970550
--- /dev/null
+++ b/control-plane/config/post-install/500-post-install-job.yaml
@@ -0,0 +1,43 @@
+# Copyright 2022 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: kafka-controller-post-install
+  namespace: knative-eventing
+  labels:
+    app: kafka-controller-post-install
+    kafka.eventing.knative.dev/release: devel
+spec:
+  ttlSecondsAfterFinished: 600
+  backoffLimit: 10
+  template:
+    metadata:
+      labels:
+        app: kafka-controller-post-install
+        kafka.eventing.knative.dev/release: devel
+      annotations:
+        sidecar.istio.io/inject: "false"
+    spec:
+      serviceAccountName: knative-kafka-controller-post-install
+      restartPolicy: OnFailure
+      containers:
+        - name: post-install
+          image: ko://knative.dev/eventing-kafka-broker/control-plane/cmd/post-install
+          env:
+            - name: SYSTEM_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
index 56daf37f5..1d6d2ed6d 100644
--- a/control-plane/pkg/prober/async_prober.go
+++ b/control-plane/pkg/prober/async_prober.go
@@ -59,7 +59,7 @@ func NewAsync(ctx context.Context, client httpClient, port string, podsLabelsSel
 	}
 }
 
-func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status {
+func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
 	address := addressable.Address
 	pods, err := a.podLister()
 	if err != nil {
@@ -98,8 +98,8 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 			With(zap.String("address", address))
 
 		currentStatus := a.cache.GetStatus(address)
-		if currentStatus == StatusReady {
-			logger.Debug("Skip probing", zap.String("status", "ready"))
+		if currentStatus == expected {
+			logger.Debug("Skip probing, status in cache is equal to the expected status", zap.String("status", currentStatus.String()))
 			wg.Done()
 			continue
 		}
@@ -112,7 +112,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 		// We want to requeue the resource only once when we have all probe request
 		// results.
 		enqueueOnce.Do(func() {
-			// Wait in a separate gorouting.
+			// Wait in a separate goroutine.
 			go func() {
 				// Wait for all the prober request results and then requeue the
 				// resource.
@@ -133,7 +133,7 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable) Status
 
 	if aggregatedCurrentKnownStatus == nil {
 		// Every status is ready, return ready.
-		return StatusReady
+		return expected
 	}
 	return *aggregatedCurrentKnownStatus
 }
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index 59679808b..857693d9c 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -40,18 +40,18 @@ type EnqueueFunc func(key types.NamespacedName)
 // Prober probes an addressable resource.
 type Prober interface {
 	// Probe probes the provided Addressable resource and returns its Status.
-	Probe(ctx context.Context, addressable Addressable) Status
+	Probe(ctx context.Context, addressable Addressable, expected Status) Status
 }
 
 // Func type is an adapter to allow the use of
 // ordinary functions as Prober. If f is a function
 // with the appropriate signature, Func(f) is a
 // Prober that calls f.
-type Func func(ctx context.Context, addressable Addressable) Status
+type Func func(ctx context.Context, addressable Addressable, expected Status) Status
 
 // Probe implements the Prober interface for Func.
-func (p Func) Probe(ctx context.Context, addressable Addressable) Status {
-	return p(ctx, addressable)
+func (p Func) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
+	return p(ctx, addressable, expected)
 }
 
 // httpClient interface is an interface for an HTTP client.
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index b54e992a4..98600a8e9 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -232,7 +232,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -299,7 +299,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 			Name:      broker.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 2365b73f7..d7d48e9b9 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -275,7 +275,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -352,7 +352,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 			Name:      channel.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 9ca81b778..14a96a558 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -237,7 +237,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		},
 	}
 
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
 		statusConditionManager.ProbesStatusNotReady(status)
 		return nil // Object will get re-queued once probe status changes.
 	}
@@ -303,7 +303,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 			Name:      ks.GetName(),
 		},
 	}
-	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
 		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
 		// for a new reconciliation.
 		return controller.NewRequeueAfter(5 * time.Second)
diff --git a/hack/artifacts-env.sh b/hack/artifacts-env.sh
index c1da54892..7b0ae2796 100755
--- a/hack/artifacts-env.sh
+++ b/hack/artifacts-env.sh
@@ -15,4 +15,6 @@ export EVENTING_KAFKA_SINK_PROMETHEUS_OPERATOR_ARTIFACT="eventing-kafka-sink-pro
 export EVENTING_KAFKA_CHANNEL_ARTIFACT="eventing-kafka-channel.yaml"
 export EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT="eventing-kafka-channel-prometheus-operator.yaml"
 
+export EVENTING_KAFKA_POST_INSTALL_ARTIFACT="eventing-kafka-post-install.yaml"
+
 export EVENTING_KAFKA_ARTIFACT="eventing-kafka.yaml"
diff --git a/hack/control-plane.sh b/hack/control-plane.sh
index 6c748dad8..52152430a 100644
--- a/hack/control-plane.sh
+++ b/hack/control-plane.sh
@@ -16,10 +16,13 @@
 
 source $(pwd)/hack/label.sh
 
-readonly CONTROL_PLANE_CONFIG_DIR=control-plane/config
+readonly CONTROL_PLANE_CONFIG_DIR=control-plane/config/eventing-kafka-broker
+readonly CONTROL_PLANE_POST_INSTALL_CONFIG_DIR=control-plane/config/post-install
 
 # Note: do not change this function name, it's used during releases.
 function control_plane_setup() {
-  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}"
+  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}" &&
+    ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_POST_INSTALL_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
+
   return $?
 }
diff --git a/hack/release.sh b/hack/release.sh
index e6a82d0ac..b47f0f00d 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -66,6 +66,7 @@ function build_release() {
     "${EVENTING_KAFKA_SINK_PROMETHEUS_OPERATOR_ARTIFACT}"
     "${EVENTING_KAFKA_CHANNEL_ARTIFACT}"
     "${EVENTING_KAFKA_CHANNEL_PROMETHEUS_OPERATOR_ARTIFACT}"
+    "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
   )
 
   # ARTIFACTS_TO_PUBLISH has to be a string, not an array.
diff --git a/hack/update-checksums.sh b/hack/update-checksums.sh
index 54f0d5d0f..df7c14e6c 100755
--- a/hack/update-checksums.sh
+++ b/hack/update-checksums.sh
@@ -26,6 +26,6 @@ fi
 
 source $(dirname $0)/../vendor/knative.dev/hack/library.sh
 
-go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-tracing.yaml
-go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-logging.yaml
-go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/200-controller/100-config-kafka-leader-election.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
index e27b94c56..d68fe4481 100644
--- a/control-plane/pkg/prober/async_prober_test.go
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -36,6 +36,7 @@ import (
 )
 
 func TestAsyncProber(t *testing.T) {
+	t.Parallel()
 
 	tt := []struct {
 		name                string
@@ -106,12 +107,13 @@ func TestAsyncProber(t *testing.T) {
 	}
 
 	for _, tc := range tt {
+		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, _ := reconcilertesting.SetupFakeContext(t)
-			ctx, cacel := context.WithCancel(ctx)
+			ctx, cancel := context.WithCancel(ctx)
 			defer func() {
 				time.Sleep(time.Second)
-				cacel()
+				cancel()
 			}()
 			wantRequestCountMin := atomic.NewInt64(int64(tc.wantRequestCountMin))
 			h := http.HandlerFunc(func(writer http.ResponseWriter, r *http.Request) {
@@ -121,7 +123,6 @@ func TestAsyncProber(t *testing.T) {
 			})
 			s := httptest.NewUnstartedServer(h)
 			s.Start()
-			time.Sleep(time.Second) // Wait for the server time to start.
 			defer s.Close()
 
 			for _, p := range tc.pods {
@@ -134,14 +135,15 @@ func TestAsyncProber(t *testing.T) {
 			prober := NewAsync(ctx, s.Client(), u.Port(), tc.podsLabelsSelector, func(key types.NamespacedName) {
 				wantRequeueCountMin.Dec()
 			})
+
 			probeFunc := func() bool {
-				status := prober.Probe(ctx, tc.addressable)
+				status := prober.Probe(ctx, tc.addressable, tc.wantStatus)
 				return status == tc.wantStatus
 			}
 
 			require.Eventuallyf(t, probeFunc, time.Second, 10*time.Millisecond, "")
-			require.GreaterOrEqual(t, int64(0), wantRequeueCountMin.Load())
-			require.GreaterOrEqual(t, int64(0), wantRequestCountMin.Load())
+			require.GreaterOrEqual(t, int64(0), wantRequestCountMin.Load(), "")
+			require.GreaterOrEqual(t, int64(0), wantRequeueCountMin.Load(), "")
 		})
 	}
 }
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
index 850bd1192..bb9bdf30b 100644
--- a/control-plane/pkg/prober/prober_test.go
+++ b/control-plane/pkg/prober/prober_test.go
@@ -28,12 +28,12 @@ func TestFuncProbe(t *testing.T) {
 
 	calls := atomic.NewInt32(0)
 	status := StatusReady
-	p := Func(func(ctx context.Context, addressable Addressable) Status {
+	p := Func(func(ctx context.Context, addressable Addressable, expected Status) Status {
 		calls.Inc()
 		return status
 	})
 
-	s := p.Probe(context.Background(), Addressable{})
+	s := p.Probe(context.Background(), Addressable{}, status)
 
 	require.Equal(t, status, s, s.String())
 	require.Equal(t, int32(1), calls.Load())
diff --git a/control-plane/pkg/prober/probertesting/mock_prober.go b/control-plane/pkg/prober/probertesting/mock_prober.go
index 0111c4850..c13110a7d 100644
--- a/control-plane/pkg/prober/probertesting/mock_prober.go
+++ b/control-plane/pkg/prober/probertesting/mock_prober.go
@@ -24,7 +24,7 @@ import (
 
 // MockProber returns a prober that always returns the provided status.
 func MockProber(status prober.Status) prober.Prober {
-	return prober.Func(func(ctx context.Context, addressable prober.Addressable) prober.Status {
+	return prober.Func(func(ctx context.Context, addressable prober.Addressable, expected prober.Status) prober.Status {
 		return status
 	})
 }
diff --git a/control-plane/pkg/prober/probertesting/mock_prober_test.go b/control-plane/pkg/prober/probertesting/mock_prober_test.go
index 560afebb0..c2c6aba91 100644
--- a/control-plane/pkg/prober/probertesting/mock_prober_test.go
+++ b/control-plane/pkg/prober/probertesting/mock_prober_test.go
@@ -53,7 +53,7 @@ func TestMockProber(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := MockProber(tt.status).Probe(tt.ctx, tt.addressable)
+			got := MockProber(tt.status).Probe(tt.ctx, tt.addressable, tt.status)
 			if diff := cmp.Diff(tt.status, got); diff != "" {
 				t.Error("(-want, got)", diff)
 			}
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 24509c2f5..4e02fbe64 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -141,6 +141,7 @@ function install_head() {
   kubectl apply -f "${EVENTING_KAFKA_BROKER_ARTIFACT}" || return $?
   kubectl apply -f "${EVENTING_KAFKA_SINK_ARTIFACT}" || return $?
   kubectl apply -f "${EVENTING_KAFKA_CHANNEL_ARTIFACT}" || return $?
+  kubectl apply -f "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" || return $?
 }
 
 function test_setup() {