fix: Use pointer types for boolean fields in PulsarFunction to prevent omitempty issues (#343)

* fix: Use pointer types for boolean fields in PulsarFunction to prevent omitempty issues

* fix ci

Author: freeznet

api/v1alpha1/pulsarfunction_types.go

@@ -38,27 +38,27 @@ type PulsarFunctionSpec struct {
 
 	// CleanupSubscription is the flag to indicate whether the subscription should be cleaned up when the function is deleted
 	// +optional
-	CleanupSubscription bool `json:"cleanupSubscription"`
+	CleanupSubscription *bool `json:"cleanupSubscription,omitempty"`
 
 	// RetainOrdering is the flag to indicate whether the function should retain ordering
 	// +optional
-	RetainOrdering bool `json:"retainOrdering"`
+	RetainOrdering *bool `json:"retainOrdering,omitempty"`
 
 	// RetainKeyOrdering is the flag to indicate whether the function should retain key ordering
 	// +optional
-	RetainKeyOrdering bool `json:"retainKeyOrdering"`
+	RetainKeyOrdering *bool `json:"retainKeyOrdering,omitempty"`
 
 	// BatchBuilder is the batch builder that the function uses
 	// +optional
 	BatchBuilder *string `json:"batchBuilder,omitempty"`
 
 	// ForwardSourceMessageProperty is the flag to indicate whether the function should forward source message properties
 	// +optional
-	ForwardSourceMessageProperty bool `json:"forwardSourceMessageProperty"`
+	ForwardSourceMessageProperty *bool `json:"forwardSourceMessageProperty,omitempty"`
 
 	// AutoAck is the flag to indicate whether the function should auto ack
 	// +optional
-	AutoAck bool `json:"autoAck"`
+	AutoAck *bool `json:"autoAck,omitempty"`
 
 	// Parallelism is the parallelism of the function
 	// +optional
@@ -186,11 +186,11 @@ type PulsarFunctionSpec struct {
 
 	// ExposePulsarAdminClientEnabled is the flag to indicate whether the function should expose pulsar admin client
 	// +optional
-	ExposePulsarAdminClientEnabled bool `json:"exposePulsarAdminClientEnabled"`
+	ExposePulsarAdminClientEnabled *bool `json:"exposePulsarAdminClientEnabled,omitempty"`
 
 	// SkipToLatest is the flag to indicate whether the function should skip to latest
 	// +optional
-	SkipToLatest bool `json:"skipToLatest"`
+	SkipToLatest *bool `json:"skipToLatest,omitempty"`
 
 	// SubscriptionPosition is the subscription position of the function
 	// +optional

api/v1alpha1/zz_generated.deepcopy.go

@@ -41,7 +41,7 @@ This table lists specifications available for the `PulsarSink` resource.
 | `topicsPattern`                | The topic pattern.                                                                                                                                    | Optional        |
 | `resources`                    | The resources of the sink.                                                                                                                            | Optional        |
 | `timeoutMs`                    | The timeout in milliseconds.                                                                                                                          | Optional        |
-| `cleanupSubscription`          | The cleanup subscription (defaults to false when omitted).                                                                                            | Optional        |
+| `cleanupSubscription`          | The cleanup subscription.                                                                                            | Optional        |
 | `retainOrdering`               | The retain ordering.                                                                                                                                  | Optional        |
 | `retainKeyOrdering`            | The retain key ordering.                                                                                                                              | Optional        |
 | `autoAck`                      | The auto ack.                                                                                                                                         | Optional        |

docs/pulsar_sink.md

@@ -46,14 +46,6 @@ const (
 	TopicDomainNonPersistent = "non-persistent"
 )
 
-// ptrBoolToBool converts a pointer to bool to bool with default value false
-func ptrBoolToBool(b *bool) bool {
-	if b == nil {
-		return false
-	}
-	return *b
-}
-
 // Type conversion functions for external library types
 
 // convertOffloadPolicies converts our local OffloadPolicies to the external library type
@@ -1281,37 +1273,58 @@ func (p *PulsarAdminClient) DeletePulsarFunction(tenant, namespace, name string)
 // ApplyPulsarFunction creates or updates a pulsar function
 func (p *PulsarAdminClient) ApplyPulsarFunction(tenant, namespace, name, packageURL string, param *v1alpha1.PulsarFunctionSpec, changed bool) error {
 	functionConfig := utils.FunctionConfig{
-		Tenant:                         tenant,
-		Namespace:                      namespace,
-		Name:                           name,
-		ClassName:                      param.ClassName,
-		Inputs:                         param.Inputs,
-		Parallelism:                    param.Parallelism,
-		TimeoutMs:                      param.TimeoutMs,
-		TopicsPattern:                  param.TopicsPattern,
-		CleanupSubscription:            param.CleanupSubscription,
-		RetainOrdering:                 param.RetainOrdering,
-		RetainKeyOrdering:              param.RetainKeyOrdering,
-		ForwardSourceMessageProperty:   param.ForwardSourceMessageProperty,
-		AutoAck:                        param.AutoAck,
-		MaxMessageRetries:              param.MaxMessageRetries,
-		CustomSerdeInputs:              param.CustomSerdeInputs,
-		CustomSchemaInputs:             param.CustomSchemaInputs,
-		InputTypeClassName:             param.InputTypeClassName,
-		Output:                         param.Output,
-		OutputSerdeClassName:           param.OutputSerdeClassName,
-		OutputSchemaType:               param.OutputSchemaType,
-		OutputTypeClassName:            param.OutputTypeClassName,
-		CustomSchemaOutputs:            param.CustomSchemaOutputs,
-		LogTopic:                       param.LogTopic,
-		ProcessingGuarantees:           param.ProcessingGuarantees,
-		DeadLetterTopic:                param.DeadLetterTopic,
-		SubName:                        param.SubName,
-		RuntimeFlags:                   param.RuntimeFlags,
-		MaxPendingAsyncRequests:        param.MaxPendingAsyncRequests,
-		ExposePulsarAdminClientEnabled: param.ExposePulsarAdminClientEnabled,
-		SkipToLatest:                   param.SkipToLatest,
-		SubscriptionPosition:           param.SubscriptionPosition,
+		Tenant:                  tenant,
+		Namespace:               namespace,
+		Name:                    name,
+		ClassName:               param.ClassName,
+		Inputs:                  param.Inputs,
+		Parallelism:             param.Parallelism,
+		TimeoutMs:               param.TimeoutMs,
+		TopicsPattern:           param.TopicsPattern,
+		MaxMessageRetries:       param.MaxMessageRetries,
+		CustomSerdeInputs:       param.CustomSerdeInputs,
+		CustomSchemaInputs:      param.CustomSchemaInputs,
+		InputTypeClassName:      param.InputTypeClassName,
+		Output:                  param.Output,
+		OutputSerdeClassName:    param.OutputSerdeClassName,
+		OutputSchemaType:        param.OutputSchemaType,
+		OutputTypeClassName:     param.OutputTypeClassName,
+		CustomSchemaOutputs:     param.CustomSchemaOutputs,
+		LogTopic:                param.LogTopic,
+		ProcessingGuarantees:    param.ProcessingGuarantees,
+		DeadLetterTopic:         param.DeadLetterTopic,
+		SubName:                 param.SubName,
+		RuntimeFlags:            param.RuntimeFlags,
+		MaxPendingAsyncRequests: param.MaxPendingAsyncRequests,
+		SubscriptionPosition:    param.SubscriptionPosition,
+	}
+
+	if param.CleanupSubscription != nil {
+		functionConfig.CleanupSubscription = *param.CleanupSubscription
+	}
+
+	if param.RetainOrdering != nil {
+		functionConfig.RetainOrdering = *param.RetainOrdering
+	}
+
+	if param.RetainKeyOrdering != nil {
+		functionConfig.RetainKeyOrdering = *param.RetainKeyOrdering
+	}
+
+	if param.ForwardSourceMessageProperty != nil {
+		functionConfig.ForwardSourceMessageProperty = *param.ForwardSourceMessageProperty
+	}
+
+	if param.AutoAck != nil {
+		functionConfig.AutoAck = *param.AutoAck
+	}
+
+	if param.ExposePulsarAdminClientEnabled != nil {
+		functionConfig.ExposePulsarAdminClientEnabled = *param.ExposePulsarAdminClientEnabled
+	}
+
+	if param.SkipToLatest != nil {
+		functionConfig.SkipToLatest = *param.SkipToLatest
 	}
 
 	if param.BatchBuilder != nil {
@@ -1460,11 +1473,7 @@ func (p *PulsarAdminClient) ApplyPulsarSink(tenant, namespace, name, packageURL
 		TopicsPattern: param.TopicsPattern,
 		TimeoutMs:     param.TimeoutMs,
 
-		CleanupSubscription: ptrBoolToBool(param.CleanupSubscription),
-		RetainOrdering:      ptrBoolToBool(param.RetainOrdering),
-		RetainKeyOrdering:   ptrBoolToBool(param.RetainKeyOrdering),
-		AutoAck:             ptrBoolToBool(param.AutoAck),
-		Parallelism:         param.Parallelism,
+		Parallelism: param.Parallelism,
 
 		SinkType: param.SinkType,
 		Archive:  packageURL,
@@ -1487,6 +1496,22 @@ func (p *PulsarAdminClient) ApplyPulsarSink(tenant, namespace, name, packageURL
 		TransformFunctionConfig:      param.TransformFunctionConfig,
 	}
 
+	if param.CleanupSubscription != nil {
+		sinkConfig.CleanupSubscription = *param.CleanupSubscription
+	}
+
+	if param.RetainOrdering != nil {
+		sinkConfig.RetainOrdering = *param.RetainOrdering
+	}
+
+	if param.RetainKeyOrdering != nil {
+		sinkConfig.RetainKeyOrdering = *param.RetainKeyOrdering
+	}
+
+	if param.AutoAck != nil {
+		sinkConfig.AutoAck = *param.AutoAck
+	}
+
 	if param.Resources != nil {
 		s, err := strconv.ParseFloat(param.Resources.CPU, 64)
 		if err != nil {

pkg/admin/impl.go

@@ -22,7 +22,6 @@ import (
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/utils/pointer"
 	"k8s.io/utils/ptr"
 
 	"github.com/streamnative/pulsar-resources-operator/api/v1alpha1"
@@ -513,15 +512,15 @@ func MakePulsarFunction(namespace, name, functionPackageUrl, connectionName stri
 			ClassName:                    "org.apache.pulsar.functions.api.examples.ExclamationFunction",
 			SubName:                      "test-sub",
 			SubscriptionPosition:         "Latest",
-			CleanupSubscription:          true,
-			SkipToLatest:                 true,
-			ForwardSourceMessageProperty: true,
-			RetainKeyOrdering:            true,
-			AutoAck:                      true,
-			MaxMessageRetries:            pointer.Int(101),
+			CleanupSubscription:          ptr.To(true),
+			SkipToLatest:                 ptr.To(true),
+			ForwardSourceMessageProperty: ptr.To(true),
+			RetainKeyOrdering:            ptr.To(true),
+			AutoAck:                      ptr.To(true),
+			MaxMessageRetries:            ptr.To(101),
 			DeadLetterTopic:              "dl-topic",
 			LogTopic:                     "func-log",
-			TimeoutMs:                    pointer.Int64(6666),
+			TimeoutMs:                    ptr.To[int64](6666),
 			Secrets: map[string]v1alpha1.FunctionSecretKeyRef{
 				"SECRET1": {
 					"sectest", "hello",