Use triggerv2 controller in shared main

control-plane/cmd/kafka-controller/main.go

@@ -27,6 +27,7 @@
 	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/signals"
 
+	"knative.dev/eventing/pkg/auth"
 	"knative.dev/eventing/pkg/eventingtls"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
@@ -38,6 +39,7 @@
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/sink"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/source"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/trigger"
+	triggerv2 "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/trigger/v2"
 )
 
 const (
@@ -64,6 +66,7 @@
 	ctx := signals.NewContext()
 	ctx = filteredFactory.WithSelectors(ctx,
 		eventingtls.TrustBundleLabelSelector,
+		auth.OIDCLabelSelector,
 	)
 	ctx = clientpool.WithKafkaClientPool(ctx)
 
@@ -81,7 +84,7 @@
 		injection.NamedControllerConstructor{
 			Name: "trigger-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return trigger.NewController(ctx, watcher, brokerEnv)
+				return triggerv2.NewController(ctx, watcher, brokerEnv)
 			},
 		},
 

control-plane/cmd/post-install/kafka_broker_deployment_deleter.go

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2024 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"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	"knative.dev/pkg/system"
+)
+
+type kafkaDeploymentDeleter struct {
+	k8s kubernetes.Interface
+}
+
+func (k *kafkaDeploymentDeleter) DeleteBrokerDeployments(ctx context.Context) error {
+	deployments := []string{
+		"kafka-broker-receiver",
+		"kafka-broker-dispatcher",
+	}
+
+	for _, deployment := range deployments {
+		if err := k.deleteDeployment(ctx, deployment); err != nil {
+			return fmt.Errorf("failed to delete deployment %s: %v", deployment, err)
+		}
+	}
+
+	return nil
+}
+
+func (k *kafkaDeploymentDeleter) deleteDeployment(ctx context.Context, deploymentName string) error {
+	err := k.waiteStatefulSetExists(ctx, deploymentName)
+	if err != nil {
+		return fmt.Errorf("failed while waiting for statefulset to come up: %w", err)
+	}
+
+	err = k.k8s.
+		AppsV1().
+		Deployments(system.Namespace()).
+		Delete(ctx, deploymentName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete deployment %s/%s: %w", system.Namespace(), deploymentName, err)
+	}
+
+	return nil
+}
+
+func (k *kafkaDeploymentDeleter) waiteStatefulSetExists(ctx context.Context, statefulSetName string) error {
+	return wait.PollUntilContextTimeout(ctx, 10*time.Second, 10*time.Minute, false, func(ctx context.Context) (done bool, err error) {
+		_, err = k.k8s.AppsV1().StatefulSets(system.Namespace()).Get(ctx, statefulSetName, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+		if err != nil {
+			return false, fmt.Errorf("failed to get statefulset %s/%s: %w", system.Namespace(), statefulSetName, err)
+		}
+		return true, nil
+	})
+}

control-plane/cmd/post-install/main.go

@@ -16,6 +16,54 @@
 
 package main
 
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+
+	"k8s.io/client-go/kubernetes"
+
+	"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)
+	}
+
+	deploymentDeleter := &kafkaDeploymentDeleter{
+		k8s: kubernetes.NewForConfigOrDie(config),
+	}
+
+	if err := deploymentDeleter.DeleteBrokerDeployments(ctx); err != nil {
+		return fmt.Errorf("broker migration failed: %v", err)
+	}
 
+	return nil
 }

control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml

@@ -168,6 +168,8 @@ rules:
       - watch
       - update
       - patch
+      - create
+      - delete
 
   # Internal APIs
   - apiGroups:
@@ -293,3 +295,4 @@ rules:
       - update
       - create
       - delete
+

control-plane/config/post-install/200-controller-cluster-role.yaml

@@ -20,4 +20,20 @@ metadata:
   name: knative-kafka-controller-post-install
   labels:
     app.kubernetes.io/version: devel
-rules: []
+rules:
+  # we need to be able to delete old deployments
+  - apiGroups:
+      - "apps"
+    resources:
+      - "deployments"
+    verbs:
+      - "delete"
+  # we need to get statefulsets
+  - apiGroups:
+      - "apps"
+    resources:
+      - "statefulsets"
+    verbs:
+      - "get"
+      - "list"
+

control-plane/pkg/kafka/clientpool/clientpool.go

@@ -154,7 +154,12 @@
 }
 
 func (cp *ClientPool) makeSaramaClient(bootstrapServers []string, secret *corev1.Secret) (sarama.Client, error) {
-	config, err := kafka.GetSaramaConfig(security.NewSaramaSecurityOptionFromSecret(secret), kafka.DisableOffsetAutoCommitConfigOption)
+	secretOpt, err := security.NewSaramaSecurityOptionFromSecret(secret)
+	if err != nil {
+		return nil, err
+	}
+
+	config, err := kafka.GetSaramaConfig(secretOpt, kafka.DisableOffsetAutoCommitConfigOption)
 	if err != nil {
 		return nil, err
 	}

control-plane/pkg/reconciler/broker/namespaced_broker.go

@@ -72,7 +72,7 @@
 	ServiceAccountLister     corelisters.ServiceAccountLister
 	ServiceLister            corelisters.ServiceLister
 	ClusterRoleBindingLister rbaclisters.ClusterRoleBindingLister
-	DeploymentLister         appslisters.DeploymentLister
+	StatefulSetLister        appslisters.StatefulSetLister
 	BrokerLister             eventinglisters.BrokerLister
 
 	// GetKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
@@ -316,7 +316,7 @@
 	}
 	resources = append(resources, additionalConfigMaps...)
 
-	additionalDeployments, err := r.deploymentsFromSystemNamespace(broker)
+	additionalDeployments, err := r.statefulSetsFromSystemNamespace(broker)
 	if err != nil {
 		return mf.Manifest{}, err
 	}
@@ -366,14 +366,14 @@
 	return mf.ManifestFrom(mf.Slice(additionalResources), mf.UseClient(r.ManifestivalClient))
 }
 
-func (r *NamespacedReconciler) deploymentsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
+func (r *NamespacedReconciler) statefulSetsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
 	deployments := []string{
 		"kafka-broker-receiver",
 		"kafka-broker-dispatcher",
 	}
 	resources := make([]unstructured.Unstructured, 0, len(deployments))
 	for _, name := range deployments {
-		resource, err := r.createManifestFromSystemDeployment(broker, name)
+		resource, err := r.createManifestFromSystemStatefulSet(broker, name)
 		if err != nil {
 			return nil, err
 		}
@@ -420,21 +420,53 @@
 	return unstructuredFromObject(cm)
 }
 
-func (r *NamespacedReconciler) createManifestFromSystemDeployment(broker *eventing.Broker, name string) (unstructured.Unstructured, error) {
-	sysDeployment, err := r.DeploymentLister.Deployments(r.SystemNamespace).Get(name)
+func (r *NamespacedReconciler) createManifestFromSystemStatefulSet(broker *eventing.Broker, name string) (unstructured.Unstructured, error) {
+	sysStatefulSet, err := r.StatefulSetLister.StatefulSets(r.SystemNamespace).Get(name)
 	if err != nil {
-		return unstructured.Unstructured{}, fmt.Errorf("failed to get Deployment %s/%s: %w", r.SystemNamespace, name, err)
+		return unstructured.Unstructured{}, fmt.Errorf("failed to get StatefulSet %s/%s: %w", r.SystemNamespace, name, err)
 	}
 
-	cm := &appsv1.Deployment{
-		TypeMeta: metav1.TypeMeta{Kind: "Deployment", APIVersion: appsv1.SchemeGroupVersion.String()},
+	spec := sysStatefulSet.Spec
+	if spec.Replicas != nil && *spec.Replicas != 1 {
+		spec.Replicas = pointer.Int32(1)
+	}
+
+	expectedVolume := corev1.Volume{
+		Name: "contract-resources",
+		VolumeSource: corev1.VolumeSource{
+			ConfigMap: &corev1.ConfigMapVolumeSource{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: "kafka-broker-brokers-triggers",
+				},
+			},
+		},
+	}
+
+	foundContractResource := false
+	for i, volume := range spec.Template.Spec.Volumes {
+		if volume.Name == "contract-resources" {
+			foundContractResource = true
+			if volume.ConfigMap == nil || volume.ConfigMap.Name != "kafka-broker-brokers-triggers" {
+				spec.Template.Spec.Volumes[i] = expectedVolume
+			}
+		}
+	}
+
+	if !foundContractResource {
+		// need to add the contract resource volume to the spec
+		spec.Template.Spec.Volumes = append(spec.Template.Spec.Volumes, expectedVolume)
+
+	}
+
+	cm := &appsv1.StatefulSet{
+		TypeMeta: metav1.TypeMeta{Kind: "StatefulSet", APIVersion: appsv1.SchemeGroupVersion.String()},
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:   broker.GetNamespace(),
-			Name:        sysDeployment.Name,
-			Labels:      sysDeployment.Labels,
-			Annotations: sysDeployment.Annotations,
+			Name:        sysStatefulSet.Name,
+			Labels:      sysStatefulSet.Labels,
+			Annotations: sysStatefulSet.Annotations,
 		},
-		Spec: sysDeployment.Spec,
+		Spec: spec,
 	}
 	return unstructuredFromObject(cm)
 }

control-plane/pkg/reconciler/broker/namespaced_broker_test.go

@@ -120,8 +120,8 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				reconcilertesting.NewDeployment("kafka-broker-receiver", SystemNamespace),
-				reconcilertesting.NewDeployment("kafka-broker-dispatcher", SystemNamespace),
+				NewStatefulSet("kafka-broker-receiver", SystemNamespace),
+				NewStatefulSet("kafka-broker-dispatcher", SystemNamespace),
 				NewServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
 				reconcilertesting.NewService("kafka-broker-ingress", SystemNamespace),
 				NewClusterRoleBinding("knative-kafka-broker-data-plane",
@@ -190,12 +190,12 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					WithNamespacedLabel,
 				),
 				ToManifestivalResource(t,
-					reconcilertesting.NewDeployment("kafka-broker-receiver", BrokerNamespace),
+					NewStatefulSet("kafka-broker-receiver", BrokerNamespace),
 					WithNamespacedBrokerOwnerRef,
 					WithNamespacedLabel,
 				),
 				ToManifestivalResource(t,
-					reconcilertesting.NewDeployment("kafka-broker-dispatcher", BrokerNamespace),
+					NewStatefulSet("kafka-broker-dispatcher", BrokerNamespace),
 					WithNamespacedBrokerOwnerRef,
 					WithNamespacedLabel,
 				),
@@ -361,8 +361,8 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 				reconcilertesting.NewConfigMap(env.DataPlaneConfigConfigMapName, SystemNamespace),
 				reconcilertesting.NewConfigMap("config-tracing", SystemNamespace),
 				reconcilertesting.NewConfigMap("kafka-config-logging", SystemNamespace),
-				reconcilertesting.NewDeployment("kafka-broker-receiver", SystemNamespace),
-				reconcilertesting.NewDeployment("kafka-broker-dispatcher", SystemNamespace),
+				NewStatefulSet("kafka-broker-receiver", SystemNamespace),
+				NewStatefulSet("kafka-broker-dispatcher", SystemNamespace),
 				NewServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
 				reconcilertesting.NewService("kafka-broker-ingress", SystemNamespace),
 				NewClusterRoleBinding("knative-kafka-broker-data-plane",
@@ -492,7 +492,7 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 			},
 			NamespaceLister:          listers.GetNamespaceLister(),
 			ConfigMapLister:          listers.GetConfigMapLister(),
-			DeploymentLister:         listers.GetDeploymentLister(),
+			StatefulSetLister:        listers.GetStatefulSetLister(),
 			BrokerLister:             listers.GetBrokerLister(),
 			ServiceAccountLister:     listers.GetServiceAccountLister(),
 			ServiceLister:            listers.GetServiceLister(),

control-plane/pkg/reconciler/broker/namespaced_controller.go

@@ -52,7 +52,7 @@
 
 	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker"
 	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
-	deploymentinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
+	statefulsetinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 	namespaceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/namespace"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
@@ -107,7 +107,7 @@
 		ServiceAccountLister:               serviceaccountinformer.Get(ctx).Lister(),
 		ServiceLister:                      serviceinformer.Get(ctx).Lister(),
 		ClusterRoleBindingLister:           clusterrolebindinginformer.Get(ctx).Lister(),
-		DeploymentLister:                   deploymentinformer.Get(ctx).Lister(),
+		StatefulSetLister:                  statefulsetinformer.Get(ctx).Lister(),
 		BrokerLister:                       brokerinformer.Get(ctx).Lister(),
 		Env:                                env,
 		Counter:                            counter.NewExpiringCounter(ctx),
@@ -173,14 +173,14 @@
 		globalResync(configMap)
 	})
 
-	deploymentinformer.Get(ctx).Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+	statefulsetinformer.Get(ctx).Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: kafka.FilterAny(
 			kafka.FilterWithLabel("app", "kafka-broker-dispatcher"),
 			kafka.FilterWithLabel("app", "kafka-broker-receiver"),
 		),
 		Handler: controller.HandleAll(controller.EnsureTypeMeta(
 			globalResync,
-			appsv1.SchemeGroupVersion.WithKind("Deployment"),
+			appsv1.SchemeGroupVersion.WithKind("StatefulSet"),
 		)),
 	})