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

api/v1alpha1/pulsarsink_types.go

@@ -42,19 +42,19 @@ type PulsarSinkSpec struct {
 
 	// CleanupSubscription is the flag to enable or disable the cleanup of subscription
 	// +optional
-	CleanupSubscription bool `json:"cleanupSubscription,omitempty"`
+	CleanupSubscription *bool `json:"cleanupSubscription,omitempty"`
 
 	// RetainOrdering is the flag to enable or disable the retain ordering
 	// +optional
-	RetainOrdering bool `json:"retainOrdering,omitempty"`
+	RetainOrdering *bool `json:"retainOrdering,omitempty"`
 
 	// RetainKeyOrdering is the flag to enable or disable the retain key ordering
 	// +optional
-	RetainKeyOrdering bool `json:"retainKeyOrdering,omitempty"`
+	RetainKeyOrdering *bool `json:"retainKeyOrdering,omitempty"`
 
 	// AutoAck is the flag to enable or disable the auto ack
 	// +optional
-	AutoAck bool `json:"autoAck,omitempty"`
+	AutoAck *bool `json:"autoAck,omitempty"`
 
 	// Parallelism is the parallelism of the PulsarSink
 	// +optional

config/samples/resource_v1alpha1_pulsarsink.yaml

@@ -20,7 +20,9 @@ metadata:
 spec:
   autoAck: true
   className: org.apache.pulsar.io.datagenerator.DataGeneratorPrintSink
-  cleanupSubscription: false
+  # Note: cleanupSubscription defaults to false when omitted
+  # Only set explicitly when you need a non-default value
+  # cleanupSubscription: false
   connectionRef:
     name: "test-pulsar-connection"
   customRuntimeOptions: {}

docs/pulsar_sink.md

@@ -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.                                                                                                                             | Optional        |
+| `cleanupSubscription`          | The cleanup subscription (defaults to false when omitted).                                                                                            | Optional        |
 | `retainOrdering`               | The retain ordering.                                                                                                                                  | Optional        |
 | `retainKeyOrdering`            | The retain key ordering.                                                                                                                              | Optional        |
 | `autoAck`                      | The auto ack.                                                                                                                                         | Optional        |

pkg/admin/impl.go

@@ -46,6 +46,14 @@ 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
@@ -1452,10 +1460,10 @@ func (p *PulsarAdminClient) ApplyPulsarSink(tenant, namespace, name, packageURL
 		TopicsPattern: param.TopicsPattern,
 		TimeoutMs:     param.TimeoutMs,
 
-		CleanupSubscription: param.CleanupSubscription,
-		RetainOrdering:      param.RetainOrdering,
-		RetainKeyOrdering:   param.RetainKeyOrdering,
-		AutoAck:             param.AutoAck,
+		CleanupSubscription: ptrBoolToBool(param.CleanupSubscription),
+		RetainOrdering:      ptrBoolToBool(param.RetainOrdering),
+		RetainKeyOrdering:   ptrBoolToBool(param.RetainKeyOrdering),
+		AutoAck:             ptrBoolToBool(param.AutoAck),
 		Parallelism:         param.Parallelism,
 
 		SinkType: param.SinkType,

tests/utils/spec.go

@@ -555,9 +555,9 @@ func MakePulsarSink(namespace, name, sinkPackageUrl, connectionName string, poli
 			Inputs:                     []string{"sink-input"},
 			Parallelism:                1,
 			ProcessingGuarantees:       "EFFECTIVELY_ONCE",
-			CleanupSubscription:        false,
+			CleanupSubscription:        ptr.To(false),
 			SourceSubscriptionPosition: "Latest",
-			AutoAck:                    true,
+			AutoAck:                    ptr.To(true),
 			ClassName:                  "org.apache.pulsar.io.datagenerator.DataGeneratorPrintSink",
 			Resources: &v1alpha1.Resources{
 				CPU:  "1",