feat: remove keda objects during reconciliation

control-plane/pkg/reconciler/consumergroup/consumergroup.go

@@ -195,6 +195,9 @@
 	} else {
 		// If KEDA is not installed or autoscaler feature disabled, do nothing
 		cg.MarkAutoscalerDisabled()
+		if err := r.deleteKedaObjects(ctx); err != nil {
+			return err
+		}
 	}
 
 	if err := r.reconcileConsumers(ctx, cg); err != nil {
@@ -845,3 +848,52 @@
 func keyOf(cg metav1.Object) string {
 	return types.NamespacedName{Namespace: cg.GetNamespace(), Name: cg.GetName()}.String()
 }
+
+// deleteKedaObjects is responsible for deleting Keda (Kubernetes-based Event Driven Autoscaling) objects, "scaled
+// objects" and "trigger authentications" from all available namespaces.
+//
+// When Keda creates a scaled object, it also generates a Horizontal Pod Autoscaler. This resource will also be removed
+// when the scaled object is deleted.
+func (r *Reconciler) deleteKedaObjects(ctx context.Context, cg *kafkainternals.ConsumerGroup) error {
+
+	// check if scaled object exist, if not ignore
+	scaledObjectName := keda.GenerateScaledObjectName(cg)
+	_, err := r.KedaClient.KedaV1alpha1().ScaledObjects(cg.Namespace).Get(ctx, scaledObjectName, metav1.GetOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get Keda scaled object: %w", err)
+	}
+	if !apierrors.IsNotFound(err) {
+		err = r.KedaClient.KedaV1alpha1().ScaledObjects(cg.Namespace).Delete(ctx, scaledObjectName, metav1.DeleteOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to delete Keda scaled object: %w", err)
+		}
+		logging.FromContext(ctx).Debugw("Keda scaled object deleted",
+			"name", scaledObjectName,
+			"namespace", cg.Namespace)
+	}
+
+	// check if trigger authentication exist , if not ignore
+	var triggerAuthName string
+	if len(cg.ObjectMeta.OwnerReferences) > 0 && cg.ObjectMeta.OwnerReferences[0].UID != "" {
+		triggerAuthName = string(cg.ObjectMeta.OwnerReferences[0].UID)
+	}
+	// skip deleting trigger authentication if object reference doesn't exist
+	if triggerAuthName == "" {
+		return nil
+	}
+
+	_, err = r.KedaClient.KedaV1alpha1().TriggerAuthentications(cg.Namespace).Get(ctx, triggerAuthName, metav1.GetOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get Keda trigger authenticatoin")
+	}
+	if !apierrors.IsNotFound(err) {
+		err = r.KedaClient.KedaV1alpha1().TriggerAuthentications(cg.Namespace).Delete(ctx, triggerAuthName, metav1.DeleteOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to delete Keda trigger authentication: %w", err)
+		}
+		logging.FromContext(ctx).Debugw("Keda trigger authentication deleted",
+			"name", triggerAuthName,
+			"namespace", cg.Namespace)
+	}
+	return nil
+}