Skip Kafka function cleanup jobs

Author: jiangpengcheng

controllers/function.go

@@ -354,8 +354,16 @@ func (r *FunctionReconciler) ApplyFunctionVPA(ctx context.Context, function *v1a
 }
 
 func (r *FunctionReconciler) ApplyFunctionCleanUpJob(ctx context.Context, function *v1alpha1.Function) error {
+	desiredJob := spec.MakeFunctionCleanUpJob(function)
+	hasCleanupFinalizer := containsCleanupFinalizer(function.ObjectMeta.Finalizers)
+	if desiredJob == nil {
+		if hasCleanupFinalizer {
+			function.ObjectMeta.Finalizers = removeCleanupFinalizer(function.ObjectMeta.Finalizers)
+			return r.Update(ctx, function)
+		}
+		return nil
+	}
 	if !spec.NeedCleanup(function) {
-		desiredJob := spec.MakeFunctionCleanUpJob(function)
 		if err := r.Delete(ctx, desiredJob, getBackgroundDeletionPolicy()); err != nil {
 			if errors.IsNotFound(err) {
 				return nil
@@ -367,12 +375,10 @@ func (r *FunctionReconciler) ApplyFunctionCleanUpJob(ctx context.Context, functi
 		}
 		return nil
 	}
-	hasCleanupFinalizer := containsCleanupFinalizer(function.ObjectMeta.Finalizers)
 	if function.Spec.CleanupSubscription {
 		// add finalizer if function is updated to clean up subscription
 		if function.ObjectMeta.DeletionTimestamp.IsZero() {
 			if !hasCleanupFinalizer {
-				desiredJob := spec.MakeFunctionCleanUpJob(function)
 				if _, err := ctrl.CreateOrUpdate(ctx, r.Client, desiredJob, func() error {
 					return nil
 				}); err != nil {
@@ -387,7 +393,6 @@ func (r *FunctionReconciler) ApplyFunctionCleanUpJob(ctx context.Context, functi
 				}
 			}
 		} else {
-			desiredJob := spec.MakeFunctionCleanUpJob(function)
 			// if function is deleting, send an "INT" signal to the cleanup job to clean up subscription
 			if hasCleanupFinalizer {
 				if err := spec.TriggerCleanup(ctx, r.Client, r.RestClient, r.Config, desiredJob); err != nil {
@@ -413,7 +418,6 @@ func (r *FunctionReconciler) ApplyFunctionCleanUpJob(ctx context.Context, functi
 				return err
 			}
 
-			desiredJob := spec.MakeFunctionCleanUpJob(function)
 			// delete the cleanup job
 			if err := r.Delete(ctx, desiredJob, getBackgroundDeletionPolicy()); err != nil {
 				return err

controllers/spec/function.go

@@ -101,6 +101,9 @@ func MakeFunctionObjectMeta(function *v1alpha1.Function) *metav1.ObjectMeta {
 }
 
 func MakeFunctionCleanUpJob(function *v1alpha1.Function) *v1.Job {
+	if isKafkaFunction(function) || function.Spec.Pulsar == nil {
+		return nil
+	}
 	labels := makeFunctionLabels(function)
 	labels["owner"] = string(function.GetUID())
 	objectMeta := &metav1.ObjectMeta{

controllers/spec/function_test.go

@@ -564,6 +564,25 @@ func TestFunctionCommandDoesNotBuildPulsarRuntimeCommandForKafkaJavaFunction(t *
 	assert.Assert(t, command == nil, "kafka messaging with java runtime should not build a pulsar command, got %v", command)
 }
 
+func TestMakeFunctionCleanUpJobSkipsKafkaFunction(t *testing.T) {
+	function := makeFunctionSample("generic-kafka-cleanup")
+	function.Spec.Messaging = v1alpha1.Messaging{
+		Kafka: &v1alpha1.KafkaMessaging{
+			BootstrapServers: "kafka:9092",
+		},
+	}
+	function.Spec.Runtime = v1alpha1.Runtime{
+		GenericRuntime: &v1alpha1.GenericRuntime{
+			FunctionFile: "/pulsar/function.py",
+			Language:     "python",
+		},
+	}
+	function.Spec.CleanupSubscription = true
+
+	cleanupJob := MakeFunctionCleanUpJob(function)
+	assert.Assert(t, cleanupJob == nil, "kafka function should not create pulsar cleanup job")
+}
+
 func TestFunctionPulsarPackageServiceDownloadFallbackToMessaging(t *testing.T) {
 	previous := utils.EnableInitContainers
 	defer func() {

pkg/webhook/validate.go

@@ -438,6 +438,10 @@ func validateFunctionMessaging(spec *v1alpha1.FunctionSpec) *field.Error {
 				spec.Kafka.AuthConfig.PlainAuthConfig.SecretName,
 				"kafka.authConfig.plainAuthConfig.secretName needs to be set")
 		}
+		if spec.CleanupSubscription {
+			return field.Invalid(field.NewPath("spec").Child("cleanupSubscription"), spec.CleanupSubscription,
+				"cleanupSubscription only supports pulsar messaging")
+		}
 		return nil
 	}
 	return validateMessaging(&spec.Messaging)

pkg/webhook/validate_test.go

@@ -62,6 +62,20 @@ func TestValidateFunctionMessagingRejectsMissingKafkaPlainAuthSecretName(t *test
 	}
 }
 
+func TestValidateFunctionMessagingRejectsKafkaCleanupSubscription(t *testing.T) {
+	err := validateFunctionMessaging(&v1alpha1.FunctionSpec{
+		CleanupSubscription: true,
+		Messaging: v1alpha1.Messaging{
+			Kafka: &v1alpha1.KafkaMessaging{
+				BootstrapServers: "kafka:9092",
+			},
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "cleanupSubscription only supports pulsar messaging") {
+		t.Fatalf("expected kafka cleanupSubscription error, got %v", err)
+	}
+}
+
 func TestValidateKafkaMessagingUnsupportedRejectsSourceAndSink(t *testing.T) {
 	messaging := &v1alpha1.Messaging{
 		Pulsar: &v1alpha1.PulsarMessaging{PulsarConfig: "pulsar-config"},