diff --git a/docs/resources/sink.md b/docs/resources/sink.md index 085d0d55..87013a93 100644 --- a/docs/resources/sink.md +++ b/docs/resources/sink.md @@ -57,11 +57,16 @@ Manages Pulsar IO sinks through the Functions Worker API. Required: -- `is_regex_pattern` (Boolean) -- `key` (String) -- `receiver_queue_size` (Number) -- `schema_type` (String) -- `serde_class_name` (String) +- `key` (String) The input topic that this consumer configuration applies to. + +Optional: + +- `consumer_properties` (Map of String) Consumer properties key/values for this topic. +- `is_regex_pattern` (Boolean) Whether the topic is a regex pattern matching multiple topics. Pulsar rejects a change to this on an existing sink. +- `pool_messages` (Boolean) Whether the consumer pools messages for this topic. +- `receiver_queue_size` (Number) The consumer receiver queue size for this topic. When omitted, the provider sends 1000, which buffers up to that many messages per sink instance. Set to 0 to disable prefetch. +- `schema_type` (String) The schema type of this topic, either a builtin schema type such as `avro` or a Schema implementation class name. Cannot be set together with `serde_class_name`. +- `serde_class_name` (String) The serde class name of this topic. Cannot be set together with `schema_type`. ## Import diff --git a/pulsar/resource_pulsar_sink.go b/pulsar/resource_pulsar_sink.go index 415b422a..5433dffd 100644 --- a/pulsar/resource_pulsar_sink.go +++ b/pulsar/resource_pulsar_sink.go @@ -21,6 +21,7 @@ import ( "context" "encoding/json" "fmt" + "sort" "strings" "github.com/apache/pulsar-client-go/pulsaradmin/pkg/rest" @@ -49,26 +50,39 @@ const ( resourceSinkInputSpecsSubsetSerdeClassNameKey = "serde_class_name" resourceSinkInputSpecsSubsetIsRegexPatternKey = "is_regex_pattern" resourceSinkInputSpecsSubsetReceiverQueueSizeKey = "receiver_queue_size" - resourceSinkProcessingGuaranteesKey = "processing_guarantees" - resourceSinkRetainOrderingKey = "retain_ordering" - resourceSinkParallelismKey = "parallelism" - resourceSinkArchiveKey = "archive" - resourceSinkClassnameKey = "classname" - resourceSinkCPUKey = "cpu" - resourceSinkRAMKey = "ram_mb" - resourceSinkDiskKey = "disk_mb" - resourceSinkConfigsKey = "configs" - resourceSinkAutoACKKey = "auto_ack" - resourceSinkTimeoutKey = "timeout_ms" - resourceSinkCustomRuntimeOptionsKey = "custom_runtime_options" - resourceSinkDeadLetterTopicKey = "dead_letter_topic" - resourceSinkMaxRedeliverCountKey = "max_redeliver_count" - resourceSinkNegativeCountRedeliveryDelayKey = "negative_ack_redelivery_delay_ms" - resourceSinkRetainKeyOrderingKey = "retain_key_ordering" - resourceSinkSinkTypeKey = "sink_type" - resourceSinkSecretsKey = "secrets" + resourceSinkInputSpecsSubsetPoolMessagesKey = "pool_messages" + //nolint:lll + resourceSinkInputSpecsSubsetConsumerPropertiesKey = "consumer_properties" + resourceSinkProcessingGuaranteesKey = "processing_guarantees" + resourceSinkRetainOrderingKey = "retain_ordering" + resourceSinkParallelismKey = "parallelism" + resourceSinkArchiveKey = "archive" + resourceSinkClassnameKey = "classname" + resourceSinkCPUKey = "cpu" + resourceSinkRAMKey = "ram_mb" + resourceSinkDiskKey = "disk_mb" + resourceSinkConfigsKey = "configs" + resourceSinkAutoACKKey = "auto_ack" + resourceSinkTimeoutKey = "timeout_ms" + resourceSinkCustomRuntimeOptionsKey = "custom_runtime_options" + resourceSinkDeadLetterTopicKey = "dead_letter_topic" + resourceSinkMaxRedeliverCountKey = "max_redeliver_count" + resourceSinkNegativeCountRedeliveryDelayKey = "negative_ack_redelivery_delay_ms" + resourceSinkRetainKeyOrderingKey = "retain_key_ordering" + resourceSinkSinkTypeKey = "sink_type" + resourceSinkSecretsKey = "secrets" ) +const defaultSinkReceiverQueueSize = 1000 + +var sinkInputSourceKeys = []string{ + resourceSinkInputsKey, + resourceSinkTopicsPatternKey, + resourceSinkCustomSerdeInputsKey, + resourceSinkCustomSchemaInputsKey, + resourceSinkInputSpecsKey, +} + var resourceSinkDescriptions = make(map[string]string) func init() { @@ -112,6 +126,7 @@ func resourcePulsarSink() *schema.Resource { ReadContext: resourcePulsarSinkRead, UpdateContext: resourcePulsarSinkUpdate, DeleteContext: resourcePulsarSinkDelete, + CustomizeDiff: resourcePulsarSinkCustomizeDiff, Description: "Manages Pulsar IO sinks through the Functions Worker API.", Importer: &schema.ResourceImporter{ StateContext: func(ctx context.Context, d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) { @@ -152,14 +167,12 @@ func resourcePulsarSink() *schema.Resource { resourceSinkInputsKey: { Type: schema.TypeSet, Optional: true, - ForceNew: true, Description: resourceSinkDescriptions[resourceSinkInputsKey], Elem: &schema.Schema{Type: schema.TypeString}, }, resourceSinkTopicsPatternKey: { Type: schema.TypeString, Optional: true, - ForceNew: true, Description: resourceSinkDescriptions[resourceSinkTopicsPatternKey], }, resourceSinkSubscriptionNameKey: { @@ -218,15 +231,57 @@ func resourcePulsarSink() *schema.Resource { resourceSinkInputSpecsKey: { Type: schema.TypeSet, Optional: true, - Computed: true, Description: resourceSinkDescriptions[resourceSinkInputSpecsKey], Elem: &schema.Resource{ Schema: map[string]*schema.Schema{ - resourceSinkInputSpecsSubsetTopicKey: {Type: schema.TypeString, Required: true}, - resourceSinkInputSpecsSubsetSchemaTypeKey: {Type: schema.TypeString, Required: true}, - resourceSinkInputSpecsSubsetSerdeClassNameKey: {Type: schema.TypeString, Required: true}, - resourceSinkInputSpecsSubsetIsRegexPatternKey: {Type: schema.TypeBool, Required: true}, - resourceSinkInputSpecsSubsetReceiverQueueSizeKey: {Type: schema.TypeInt, Required: true}, + resourceSinkInputSpecsSubsetTopicKey: { + Type: schema.TypeString, + Required: true, + Description: "The input topic that this consumer configuration applies to.", + }, + resourceSinkInputSpecsSubsetSchemaTypeKey: { + Type: schema.TypeString, + Optional: true, + //nolint:lll + Description: "The schema type of this topic, either a builtin schema type such as `avro` or a Schema implementation class name. Cannot be set together with `serde_class_name`.", + }, + resourceSinkInputSpecsSubsetSerdeClassNameKey: { + Type: schema.TypeString, + Optional: true, + Description: "The serde class name of this topic. Cannot be set together with `schema_type`.", + }, + resourceSinkInputSpecsSubsetIsRegexPatternKey: { + Type: schema.TypeBool, + Optional: true, + //nolint:lll + Description: "Whether the topic is a regex pattern matching multiple topics. Pulsar rejects a change to this on an existing sink.", + }, + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: { + Type: schema.TypeInt, + Optional: true, + Default: defaultSinkReceiverQueueSize, + //nolint:lll + Description: "The consumer receiver queue size for this topic. When omitted, the provider sends 1000, which buffers up to that many messages per sink instance. Set to 0 to disable prefetch.", + ValidateFunc: func(val interface{}, key string) ([]string, []error) { + if v := val.(int); v < 0 { + return nil, []error{ + fmt.Errorf("%s must be greater than or equal to 0, got %d", key, v), + } + } + return nil, nil + }, + }, + resourceSinkInputSpecsSubsetPoolMessagesKey: { + Type: schema.TypeBool, + Optional: true, + Description: "Whether the consumer pools messages for this topic.", + }, + resourceSinkInputSpecsSubsetConsumerPropertiesKey: { + Type: schema.TypeMap, + Optional: true, + Description: "Consumer properties key/values for this topic.", + Elem: &schema.Schema{Type: schema.TypeString}, + }, }, }, }, @@ -399,21 +454,10 @@ func resourcePulsarSinkRead(ctx context.Context, d *schema.ResourceData, meta in return diag.FromErr(errors.Wrapf(err, "failed to get %s sink from %s/%s", name, tenant, namespace)) } - inputs := make([]string, len(sinkConfig.Inputs)) - copy(inputs, sinkConfig.Inputs) - - err = d.Set(resourceSinkInputsKey, inputs) - if err != nil { + if err = unmarshalSinkInputSpecs(sinkConfig, d); err != nil { return diag.FromErr(err) } - if sinkConfig.TopicsPattern != nil { - err = d.Set(resourceSinkTopicsPatternKey, sinkConfig.TopicsPattern) - if err != nil { - return diag.FromErr(err) - } - } - if len(sinkConfig.SourceSubscriptionName) != 0 { err = d.Set(resourceSinkSubscriptionNameKey, sinkConfig.SourceSubscriptionName) if err != nil { @@ -426,46 +470,6 @@ func resourcePulsarSinkRead(ctx context.Context, d *schema.ResourceData, meta in return diag.FromErr(err) } - if len(sinkConfig.TopicToSerdeClassName) != 0 { - customSerdeInputs := make(map[string]interface{}, len(sinkConfig.TopicToSerdeClassName)) - for key, value := range sinkConfig.TopicToSerdeClassName { - customSerdeInputs[key] = value - } - err = d.Set(resourceSinkCustomSerdeInputsKey, customSerdeInputs) - if err != nil { - return diag.FromErr(err) - } - } - - if len(sinkConfig.TopicToSchemaType) != 0 { - customSchemaInputs := make(map[string]interface{}, len(sinkConfig.TopicToSchemaType)) - for key, value := range sinkConfig.TopicToSchemaType { - customSchemaInputs[key] = value - } - - err = d.Set(resourceSinkCustomSchemaInputsKey, customSchemaInputs) - if err != nil { - return diag.FromErr(err) - } - } - - if len(sinkConfig.InputSpecs) > 0 { - var inputSpecs []interface{} - for key, config := range sinkConfig.InputSpecs { - item := make(map[string]interface{}) - item[resourceSinkInputSpecsSubsetTopicKey] = key - item[resourceSinkInputSpecsSubsetSchemaTypeKey] = config.SchemaType - item[resourceSinkInputSpecsSubsetSerdeClassNameKey] = config.SerdeClassName - item[resourceSinkInputSpecsSubsetIsRegexPatternKey] = config.RegexPattern - item[resourceSinkInputSpecsSubsetReceiverQueueSizeKey] = config.ReceiverQueueSize - inputSpecs = append(inputSpecs, item) - } - err = d.Set(resourceSinkInputSpecsKey, inputSpecs) - if err != nil { - return diag.FromErr(err) - } - } - err = d.Set(resourceSinkParallelismKey, sinkConfig.Parallelism) if err != nil { return diag.FromErr(err) @@ -634,6 +638,559 @@ func resourcePulsarSinkDelete(ctx context.Context, d *schema.ResourceData, meta return diag.FromErr(client.DeleteSink(tenant, namespace, name)) } +// resourcePulsarSinkCustomizeDiff validates input_specs and mirrors the broker's update rules: +// consumer settings can change in place, while changing a topic or its regex flag replaces the sink. +// Moving an unchanged topic between a legacy input field and input_specs remains an in-place update. +func resourcePulsarSinkCustomizeDiff(_ context.Context, diff *schema.ResourceDiff, _ interface{}) error { + newSpecs := diff.Get(resourceSinkInputSpecsKey) + if err := validateSinkInputSpecs(newSpecs); err != nil { + return err + } + + if diff.Id() == "" { + return nil + } + + inputChanged := false + for _, key := range sinkInputSourceKeys { + if diff.HasChange(key) { + inputChanged = true + break + } + } + if !inputChanged { + return nil + } + + oldInputs, newInputs := diff.GetChange(resourceSinkInputsKey) + oldPattern, newPattern := diff.GetChange(resourceSinkTopicsPatternKey) + oldCustomSerde, newCustomSerde := diff.GetChange(resourceSinkCustomSerdeInputsKey) + oldCustomSchema, newCustomSchema := diff.GetChange(resourceSinkCustomSchemaInputsKey) + oldSpecs, newSpecs := diff.GetChange(resourceSinkInputSpecsKey) + + oldTopics := effectiveSinkInputTopics( + oldInputs, oldPattern, oldCustomSerde, oldCustomSchema, oldSpecs, + ) + newTopics := effectiveSinkInputTopics( + newInputs, newPattern, newCustomSerde, newCustomSchema, newSpecs, + ) + + if len(oldTopics) != len(newTopics) { + return forceNewSinkInputTopology(diff, oldSpecs, newSpecs) + } + for topic, regexPattern := range newTopics { + oldRegexPattern, ok := oldTopics[topic] + if !ok || oldRegexPattern != regexPattern { + return forceNewSinkInputTopology(diff, oldSpecs, newSpecs) + } + } + + return nil +} + +func forceNewSinkInputTopology(diff *schema.ResourceDiff, oldSpecs, newSpecs interface{}) error { + if diff.HasChange(resourceSinkInputSpecsKey) { + return forceNewSinkInputSpecs(diff, oldSpecs, newSpecs) + } + + if diff.HasChange(resourceSinkInputsKey) { + return forceNewSinkInputSet(diff, resourceSinkInputsKey) + } + + if diff.HasChange(resourceSinkTopicsPatternKey) { + return diff.ForceNew(resourceSinkTopicsPatternKey) + } + + for _, key := range []string{ + resourceSinkCustomSerdeInputsKey, + resourceSinkCustomSchemaInputsKey, + } { + if diff.HasChange(key) { + return forceNewSinkInputMap(diff, key) + } + } + + return errors.New("input topology changed without an input attribute diff") +} + +func forceNewSinkInputSet(diff *schema.ResourceDiff, key string) error { + if err := diff.ForceNew(key); err != nil { + return err + } + + oldValue, newValue := diff.GetChange(key) + for _, value := range []interface{}{oldValue, newValue} { + set, ok := value.(*schema.Set) + if !ok { + continue + } + for _, item := range set.List() { + itemKey := fmt.Sprintf("%s.%d", key, set.F(item)) + if diff.HasChange(itemKey) { + // The aggregate set is already ForceNew; one changed element is sufficient to + // preserve that decision when the SDK rehashes an element. + return diff.ForceNew(itemKey) + } + } + } + + return nil +} + +func forceNewSinkInputMap(diff *schema.ResourceDiff, key string) error { + if err := diff.ForceNew(key); err != nil { + return err + } + + oldValue, newValue := diff.GetChange(key) + mapKeys := sinkInputMapKeys(oldValue) + for topic := range sinkInputMapKeys(newValue) { + mapKeys[topic] = true + } + for topic := range mapKeys { + itemKey := key + "." + topic + if diff.HasChange(itemKey) { + // The aggregate map is already ForceNew; one changed entry is sufficient to + // preserve that decision in the flattened diff. + return diff.ForceNew(itemKey) + } + } + + return nil +} + +// A set-level ForceNew is insufficient when a set element changes but the element count does not. +// Mark the nested topology attribute too so Terraform preserves the replacement decision. +func forceNewSinkInputSpecs(diff *schema.ResourceDiff, oldSpecs, newSpecs interface{}) error { + if err := diff.ForceNew(resourceSinkInputSpecsKey); err != nil { + return err + } + + for _, attribute := range []string{ + resourceSinkInputSpecsSubsetTopicKey, + resourceSinkInputSpecsSubsetIsRegexPatternKey, + } { + for _, specs := range []interface{}{oldSpecs, newSpecs} { + set, ok := specs.(*schema.Set) + if !ok { + continue + } + + for _, item := range set.List() { + key := fmt.Sprintf("%s.%d.%s", resourceSinkInputSpecsKey, set.F(item), attribute) + if !diff.HasChange(key) { + continue + } + if err := diff.ForceNew(key); err != nil { + return err + } + } + } + } + + return nil +} + +// effectiveSinkInputTopics maps every input topic to its regex flag in the broker's create-path +// precedence order. All representations share the broker's inputSpecs keyspace, so an identical +// topic/pattern string follows the same last-write-wins behavior as SinkConfigUtils. input_specs is +// applied last and is therefore the canonical representation. +func effectiveSinkInputTopics( + inputs, topicsPattern, customSerdeInputs, customSchemaInputs, inputSpecs interface{}, +) map[string]bool { + topics := map[string]bool{} + + if set, ok := inputs.(*schema.Set); ok { + for _, item := range set.List() { + if topic, ok := item.(string); ok && topic != "" { + topics[topic] = false + } + } + } + + if pattern, ok := topicsPattern.(string); ok && pattern != "" { + topics[pattern] = true + } + + for topic := range sinkInputMapKeys(customSerdeInputs) { + topics[topic] = false + } + for topic := range sinkInputMapKeys(customSchemaInputs) { + topics[topic] = false + } + + for topic, consumerConfig := range sinkInputSpecsFromSchema(inputSpecs) { + topics[topic] = consumerConfig.RegexPattern + } + + return topics +} + +func sinkInputMapKeys(value interface{}) map[string]bool { + keys := map[string]bool{} + + switch values := value.(type) { + case map[string]interface{}: + for key := range values { + if key != "" { + keys[key] = true + } + } + case map[string]string: + for key := range values { + if key != "" { + keys[key] = true + } + } + } + + return keys +} + +// sinkStringMap narrows a schema.TypeMap value to map[string]string, returning nil when empty so +// the field is omitted from the request payload. +func sinkStringMap(value interface{}) map[string]string { + interMap, ok := value.(map[string]interface{}) + if !ok || len(interMap) == 0 { + return nil + } + + stringMap := make(map[string]string, len(interMap)) + for key, item := range interMap { + stringMap[key], _ = item.(string) + } + + return stringMap +} + +func sinkInputSpecsFromSchema(inputSpecs interface{}) map[string]utils.ConsumerConfig { + set, ok := inputSpecs.(*schema.Set) + if !ok || set.Len() == 0 { + return nil + } + + specs := make(map[string]utils.ConsumerConfig, set.Len()) + for _, item := range set.List() { + spec, ok := item.(map[string]interface{}) + if !ok { + continue + } + + topic, _ := spec[resourceSinkInputSpecsSubsetTopicKey].(string) + if topic == "" { + continue + } + + consumerConfig := utils.ConsumerConfig{} + if v, ok := spec[resourceSinkInputSpecsSubsetSchemaTypeKey].(string); ok { + consumerConfig.SchemaType = v + } + if v, ok := spec[resourceSinkInputSpecsSubsetSerdeClassNameKey].(string); ok { + consumerConfig.SerdeClassName = v + } + if v, ok := spec[resourceSinkInputSpecsSubsetIsRegexPatternKey].(bool); ok { + consumerConfig.RegexPattern = v + } + if v, ok := spec[resourceSinkInputSpecsSubsetReceiverQueueSizeKey].(int); ok { + consumerConfig.SetReceiverQueueSize(v) + } + if v, ok := spec[resourceSinkInputSpecsSubsetPoolMessagesKey].(bool); ok { + consumerConfig.PoolMessages = v + } + consumerConfig.ConsumerProperties = sinkStringMap( + spec[resourceSinkInputSpecsSubsetConsumerPropertiesKey]) + + specs[topic] = consumerConfig + } + + if len(specs) == 0 { + return nil + } + + return specs +} + +func sinkLegacyInputMap( + value interface{}, inputSpecs map[string]utils.ConsumerConfig, +) map[string]string { + stringMap := sinkStringMap(value) + for topic := range inputSpecs { + delete(stringMap, topic) + } + if len(stringMap) == 0 { + return nil + } + + return stringMap +} + +// unmarshalSinkInputSpecs keeps the input representation chosen in configuration while refreshing +// its values from the broker's canonical InputSpecs map. Pulsar returns every sink input through both +// Inputs and InputSpecs and does not reconstruct the legacy pattern or custom maps, so copying the +// response verbatim would invent state and cause replacement drift. +func unmarshalSinkInputSpecs(sinkConfig utils.SinkConfig, d *schema.ResourceData) error { + remoteSpecs := sinkConfig.InputSpecs + if remoteSpecs == nil { + remoteSpecs = map[string]utils.ConsumerConfig{} + } + // Current Pulsar versions return every entry through InputSpecs. Retain a defensive fallback + // for older or partial responses that expose a plain input only through Inputs. + for _, topic := range sinkConfig.Inputs { + if _, ok := remoteSpecs[topic]; !ok { + remoteSpecs[topic] = utils.ConsumerConfig{} + } + } + + declared := sinkInputSpecsFromSchema(d.Get(resourceSinkInputSpecsKey)) + if !hasConfiguredSinkInputs(d, declared) { + return unmarshalImportedSinkInputs(remoteSpecs, d) + } + + covered, err := refreshSinkLegacyInputs(remoteSpecs, declared, d) + if err != nil { + return err + } + + specs := make([]interface{}, 0, len(remoteSpecs)) + for topic, consumerConfig := range remoteSpecs { + _, isDeclared := declared[topic] + if covered[topic] && !isDeclared { + continue + } + specs = append(specs, flattenSinkInputSpec(topic, consumerConfig)) + } + + return d.Set(resourceSinkInputSpecsKey, specs) +} + +func hasConfiguredSinkInputs(d *schema.ResourceData, declared map[string]utils.ConsumerConfig) bool { + if len(declared) != 0 { + return true + } + for _, key := range []string{ + resourceSinkInputsKey, + resourceSinkTopicsPatternKey, + resourceSinkCustomSerdeInputsKey, + resourceSinkCustomSchemaInputsKey, + } { + if _, ok := d.GetOk(key); ok { + return true + } + } + return false +} + +func refreshSinkLegacyInputs( + remoteSpecs, declared map[string]utils.ConsumerConfig, d *schema.ResourceData, +) (map[string]bool, error) { + covered := map[string]bool{} + + if inter, ok := d.GetOk(resourceSinkInputsKey); ok { + inputs := make([]string, 0, inter.(*schema.Set).Len()) + for _, item := range inter.(*schema.Set).List() { + topic := item.(string) + remote, exists := remoteSpecs[topic] + _, isDeclared := declared[topic] + if !exists || (!isDeclared && remote.RegexPattern) { + continue + } + inputs = append(inputs, topic) + covered[topic] = true + } + if err := d.Set(resourceSinkInputsKey, inputs); err != nil { + return nil, err + } + } + + if inter, ok := d.GetOk(resourceSinkTopicsPatternKey); ok { + pattern := inter.(string) + remote, exists := remoteSpecs[pattern] + _, isDeclared := declared[pattern] + if exists && (isDeclared || remote.RegexPattern) { + covered[pattern] = true + } else if err := d.Set(resourceSinkTopicsPatternKey, ""); err != nil { + return nil, err + } + } + + type legacyMap struct { + key string + value func(utils.ConsumerConfig) string + } + for _, legacy := range []legacyMap{ + {resourceSinkCustomSerdeInputsKey, func(config utils.ConsumerConfig) string { + return config.SerdeClassName + }}, + {resourceSinkCustomSchemaInputsKey, func(config utils.ConsumerConfig) string { + return config.SchemaType + }}, + } { + inter, ok := d.GetOk(legacy.key) + if !ok { + continue + } + values := sinkStringMap(inter) + refreshed := make(map[string]string, len(values)) + for topic, value := range values { + remote, exists := remoteSpecs[topic] + _, isDeclared := declared[topic] + if !exists { + continue + } + if isDeclared { + // input_specs wins on the wire, so preserve an overlapped legacy value that the + // broker cannot reconstruct independently. + refreshed[topic] = value + covered[topic] = true + continue + } + if remote.RegexPattern { + continue + } + if remoteValue := legacy.value(remote); remoteValue != "" { + refreshed[topic] = remoteValue + covered[topic] = true + } + } + if err := d.Set(legacy.key, refreshed); err != nil { + return nil, err + } + } + + return covered, nil +} + +func unmarshalImportedSinkInputs( + remoteSpecs map[string]utils.ConsumerConfig, d *schema.ResourceData, +) error { + topics := make([]string, 0, len(remoteSpecs)) + regexCandidates := 0 + for topic, consumerConfig := range remoteSpecs { + topics = append(topics, topic) + if sinkInputSpecUsesOnlyLegacyDefaults(consumerConfig) && consumerConfig.RegexPattern && + consumerConfig.SerdeClassName == "" && consumerConfig.SchemaType == "" { + regexCandidates++ + } + } + sort.Strings(topics) + + inputs := make([]string, 0, len(remoteSpecs)) + customSerdeInputs := map[string]string{} + customSchemaInputs := map[string]string{} + pattern := "" + specs := make([]interface{}, 0, len(remoteSpecs)) + for _, topic := range topics { + consumerConfig := remoteSpecs[topic] + if sinkInputSpecUsesOnlyLegacyDefaults(consumerConfig) { + switch { + case consumerConfig.RegexPattern && regexCandidates == 1 && + consumerConfig.SerdeClassName == "" && consumerConfig.SchemaType == "": + pattern = topic + continue + case !consumerConfig.RegexPattern && consumerConfig.SerdeClassName != "": + customSerdeInputs[topic] = consumerConfig.SerdeClassName + continue + case !consumerConfig.RegexPattern && consumerConfig.SchemaType != "": + customSchemaInputs[topic] = consumerConfig.SchemaType + continue + case !consumerConfig.RegexPattern && consumerConfig.SerdeClassName == "" && + consumerConfig.SchemaType == "": + inputs = append(inputs, topic) + continue + } + } + specs = append(specs, flattenSinkInputSpec(topic, consumerConfig)) + } + + for key, value := range map[string]interface{}{ + resourceSinkInputsKey: inputs, + resourceSinkTopicsPatternKey: pattern, + resourceSinkCustomSerdeInputsKey: customSerdeInputs, + resourceSinkCustomSchemaInputsKey: customSchemaInputs, + resourceSinkInputSpecsKey: specs, + } { + if err := d.Set(key, value); err != nil { + return err + } + } + return nil +} + +func sinkInputSpecUsesOnlyLegacyDefaults(consumerConfig utils.ConsumerConfig) bool { + return !consumerConfig.HasReceiverQueueSize() && + !consumerConfig.PoolMessages && + len(consumerConfig.ConsumerProperties) == 0 && + len(consumerConfig.SchemaProperties) == 0 && + consumerConfig.CryptoConfig == nil && + (consumerConfig.SchemaType == "" || consumerConfig.SerdeClassName == "") +} + +func flattenSinkInputSpec(topic string, consumerConfig utils.ConsumerConfig) map[string]interface{} { + spec := map[string]interface{}{ + resourceSinkInputSpecsSubsetTopicKey: topic, + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: defaultSinkReceiverQueueSize, + resourceSinkInputSpecsSubsetIsRegexPatternKey: consumerConfig.RegexPattern, + resourceSinkInputSpecsSubsetPoolMessagesKey: consumerConfig.PoolMessages, + } + + if consumerConfig.HasReceiverQueueSize() { + spec[resourceSinkInputSpecsSubsetReceiverQueueSizeKey] = consumerConfig.ReceiverQueueSize + } + if consumerConfig.SchemaType != "" { + spec[resourceSinkInputSpecsSubsetSchemaTypeKey] = consumerConfig.SchemaType + } + if consumerConfig.SerdeClassName != "" { + spec[resourceSinkInputSpecsSubsetSerdeClassNameKey] = consumerConfig.SerdeClassName + } + if len(consumerConfig.ConsumerProperties) != 0 { + spec[resourceSinkInputSpecsSubsetConsumerPropertiesKey] = + convertToInterfaceMap(consumerConfig.ConsumerProperties) + } + + return spec +} + +// validateSinkInputSpecs enforces the two rules Pulsar applies to inputSpecs that the schema cannot +// express: topics are the map key so they must be unique, and SinkConfigUtils rejects a spec that +// sets both schemaType and serdeClassName. +func validateSinkInputSpecs(inputSpecs interface{}) error { + set, ok := inputSpecs.(*schema.Set) + if !ok || set.Len() == 0 { + return nil + } + + seenTopics := make(map[string]bool, set.Len()) + for _, item := range set.List() { + spec, ok := item.(map[string]interface{}) + if !ok { + continue + } + + topic, _ := spec[resourceSinkInputSpecsSubsetTopicKey].(string) + if topic == "" { + // The SDK can include an empty placeholder while diffing TypeSet elements. The nested + // Required schema still validates actual user configuration. + continue + } + if seenTopics[topic] { + return fmt.Errorf("%s contains duplicate %s %q", + resourceSinkInputSpecsKey, resourceSinkInputSpecsSubsetTopicKey, topic) + } + seenTopics[topic] = true + + schemaType, _ := spec[resourceSinkInputSpecsSubsetSchemaTypeKey].(string) + serdeClassName, _ := spec[resourceSinkInputSpecsSubsetSerdeClassNameKey].(string) + if schemaType != "" && serdeClassName != "" { + return fmt.Errorf("%s %q cannot set both %s and %s", + resourceSinkInputSpecsKey, topic, + resourceSinkInputSpecsSubsetSchemaTypeKey, + resourceSinkInputSpecsSubsetSerdeClassNameKey) + } + } + + return nil +} + func marshalSinkConfig(d *schema.ResourceData) (*utils.SinkConfig, error) { sinkConfig := &utils.SinkConfig{} @@ -649,20 +1206,37 @@ func marshalSinkConfig(d *schema.ResourceData) (*utils.SinkConfig, error) { sinkConfig.Name = inter.(string) } + if err := validateSinkInputSpecs(d.Get(resourceSinkInputSpecsKey)); err != nil { + return nil, err + } + + inputSpecs := sinkInputSpecsFromSchema(d.Get(resourceSinkInputSpecsKey)) + if len(inputSpecs) != 0 { + sinkConfig.InputSpecs = inputSpecs + } + if inter, ok := d.GetOk(resourceSinkInputsKey); ok { inputsSet := inter.(*schema.Set) var inputs []string for _, item := range inputsSet.List() { - inputs = append(inputs, item.(string)) + topic := item.(string) + if _, isInputSpec := inputSpecs[topic]; isInputSpec { + continue + } + inputs = append(inputs, topic) } - sinkConfig.Inputs = inputs + if len(inputs) != 0 { + sinkConfig.Inputs = inputs + } } if inter, ok := d.GetOk(resourceSinkTopicsPatternKey); ok { pattern := inter.(string) - sinkConfig.TopicsPattern = &pattern + if _, isInputSpec := inputSpecs[pattern]; !isInputSpec { + sinkConfig.TopicsPattern = &pattern + } } if inter, ok := d.GetOk(resourceSinkSubscriptionNameKey); ok { @@ -678,43 +1252,11 @@ func marshalSinkConfig(d *schema.ResourceData) (*utils.SinkConfig, error) { } if inter, ok := d.GetOk(resourceSinkCustomSerdeInputsKey); ok { - interMap := inter.(map[string]interface{}) - stringMap := make(map[string]string, len(interMap)) - - for key, value := range interMap { - stringMap[key] = value.(string) - } - - sinkConfig.TopicToSerdeClassName = stringMap + sinkConfig.TopicToSerdeClassName = sinkLegacyInputMap(inter, inputSpecs) } if inter, ok := d.GetOk(resourceSinkCustomSchemaInputsKey); ok { - interMap := inter.(map[string]interface{}) - stringMap := make(map[string]string, len(interMap)) - - for key, value := range interMap { - stringMap[key] = value.(string) - } - - sinkConfig.TopicToSchemaType = stringMap - } - - if inter, ok := d.GetOk(resourceSinkInputSpecsKey); ok { - set := inter.(*schema.Set) - if set.Len() > 0 { - inputSpecs := make(map[string]utils.ConsumerConfig) - for _, n := range set.List() { - m := n.(map[string]interface{}) - inputSpec := utils.ConsumerConfig{ - SchemaType: m[resourceSinkInputSpecsSubsetSchemaTypeKey].(string), - SerdeClassName: m[resourceSinkInputSpecsSubsetSerdeClassNameKey].(string), - RegexPattern: m[resourceSinkInputSpecsSubsetIsRegexPatternKey].(bool), - ReceiverQueueSize: m[resourceSinkInputSpecsSubsetReceiverQueueSizeKey].(int), - } - inputSpecs[m[resourceSinkInputSpecsSubsetTopicKey].(string)] = inputSpec - } - sinkConfig.InputSpecs = inputSpecs - } + sinkConfig.TopicToSchemaType = sinkLegacyInputMap(inter, inputSpecs) } if inter, ok := d.GetOk(resourceSinkProcessingGuaranteesKey); ok { diff --git a/pulsar/resource_pulsar_sink_test.go b/pulsar/resource_pulsar_sink_test.go index 55752e33..2c2ea0c4 100644 --- a/pulsar/resource_pulsar_sink_test.go +++ b/pulsar/resource_pulsar_sink_test.go @@ -70,11 +70,22 @@ func TestSink(t *testing.T) { return errors.New("resource id should be tenant/namespace/name format") } - _, err := client.GetSink(parts[0], parts[1], parts[2]) + sinkConfig, err := client.GetSink(parts[0], parts[1], parts[2]) if err != nil { return err } + inputSpec := sinkConfig.InputSpecs["sink-1-topic"] + if !inputSpec.HasReceiverQueueSize() || inputSpec.ReceiverQueueSize != 0 { + return fmt.Errorf("receiver_queue_size=0 did not round-trip: %#v", inputSpec) + } + if !inputSpec.PoolMessages { + return fmt.Errorf("pool_messages did not round-trip: %#v", inputSpec) + } + if inputSpec.ConsumerProperties["application"] != "billing" { + return fmt.Errorf("consumer_properties did not round-trip: %#v", inputSpec) + } + return nil }), }, @@ -137,11 +148,17 @@ func TestImportExistingSink(t *testing.T) { CheckDestroy: testPulsarSinkDestroy, Steps: []resource.TestStep{ { - ResourceName: "pulsar_sink.test", - ImportState: true, - Config: testSampleSink(sinkName), - ImportStateId: fmt.Sprintf("public/default/%s", sinkName), - ImportStateCheck: testSinkImported(), + ResourceName: "pulsar_sink.test", + ImportState: true, + Config: testSampleSink(sinkName), + ImportStateId: fmt.Sprintf("public/default/%s", sinkName), + ImportStateCheck: testSinkImported(), + ImportStatePersist: true, + }, + { + Config: testSampleSink(sinkName), + PlanOnly: true, + ExpectNonEmptyPlan: false, }, }, }) @@ -153,8 +170,23 @@ func testSinkImported() resource.ImportStateCheckFunc { return fmt.Errorf("expected %d states, got %d: %#v", 1, len(s), s) } - if len(s[0].Attributes) != 30 { - return fmt.Errorf("expected %d attrs, got %d: %#v", 30, len(s[0].Attributes), s[0].Attributes) + attributes := s[0].Attributes + if attributes[resourceSinkInputsKey+".#"] != "1" { + return fmt.Errorf("expected one imported legacy input, got %#v", attributes) + } + foundTopic := false + for key, value := range attributes { + if strings.HasPrefix(key, resourceSinkInputsKey+".") && + key != resourceSinkInputsKey+".#" && value == "sink-1-topic" { + foundTopic = true + break + } + } + if !foundTopic { + return fmt.Errorf("imported input topic is missing: %#v", attributes) + } + if count := attributes[resourceSinkInputSpecsKey+".#"]; count != "" && count != "0" { + return fmt.Errorf("plain imported input should not invent input_specs: %#v", attributes) } return nil @@ -234,7 +266,7 @@ resource "pulsar_sink" "test" { negative_ack_redelivery_delay_ms = 3000 retain_key_ordering = false retain_ordering = true - secrets ="{\"SECRET1\": {\"path\": \"sectest\", \"key\": \"hello\"}}" + secrets ="{\"secret1\": {\"path\": \"sectest\", \"key\": \"hello\"}}" processing_guarantees = "EFFECTIVELY_ONCE" @@ -244,6 +276,12 @@ resource "pulsar_sink" "test" { archive = "%s" configs = "{\"jdbcUrl\":\"jdbc:postgresql://localhost:5432/pulsar_postgres_jdbc_sink\",\"password\":\"password\",\"tableName\":\"pulsar_postgres_jdbc_sink\",\"userName\":\"postgres\"}" + + # Pulsar does not return the original package URL from GET, and secrets are normalized when read. + # Ignore those unrelated values while checking that imported input state plans cleanly. + lifecycle { + ignore_changes = [archive, secrets] + } } `, name, testdataArchive) } @@ -256,6 +294,9 @@ func TestSinkUpdate(t *testing.T) { configString := string(configBytes) newName := "sink" + acctest.RandString(10) configString = strings.ReplaceAll(configString, "sink-1", newName) + updatedConfigString := strings.Replace(configString, + "receiver_queue_size = 0", "receiver_queue_size = 100", 1) + var createdID string resource.Test(t, resource.TestCase{ PreCheck: func() { testAccPreCheck(t) }, @@ -283,12 +324,44 @@ func TestSinkUpdate(t *testing.T) { if err != nil { return err } + createdID = rs.Primary.ID return nil }), }, { - Config: configString, + Config: updatedConfigString, + Check: resource.ComposeTestCheckFunc(func(s *terraform.State) error { + name := "pulsar_sink." + newName + rs, ok := s.RootModule().Resources[name] + if !ok { + return fmt.Errorf("%s not be found", name) + } + if rs.Primary.ID != createdID { + return fmt.Errorf("sink was replaced: id changed from %s to %s", createdID, rs.Primary.ID) + } + + parts := strings.Split(rs.Primary.ID, "/") + if len(parts) != 3 { + return errors.New("resource id should be tenant/namespace/name format") + } + sinkConfig, err := getV3ClientFromMeta(testAccProvider.Meta()).Sinks().GetSink( + parts[0], parts[1], parts[2]) + if err != nil { + return err + } + inputSpec := sinkConfig.InputSpecs[newName+"-topic"] + if !inputSpec.HasReceiverQueueSize() || inputSpec.ReceiverQueueSize != 100 { + return fmt.Errorf("receiver queue size update did not round-trip: %#v", inputSpec) + } + if inputSpec.ConsumerProperties["application"] != "billing" || !inputSpec.PoolMessages { + return fmt.Errorf("input spec properties were lost during update: %#v", inputSpec) + } + return nil + }), + }, + { + Config: updatedConfigString, PlanOnly: true, ExpectNonEmptyPlan: false, }, diff --git a/pulsar/resource_pulsar_sink_unit_test.go b/pulsar/resource_pulsar_sink_unit_test.go new file mode 100644 index 00000000..f4740471 --- /dev/null +++ b/pulsar/resource_pulsar_sink_unit_test.go @@ -0,0 +1,526 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 pulsar + +import ( + "context" + "testing" + + "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils" + "github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema" + "github.com/hashicorp/terraform-plugin-sdk/v2/terraform" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func sinkInputSpec(topic string, overrides map[string]interface{}) map[string]interface{} { + spec := map[string]interface{}{ + resourceSinkInputSpecsSubsetTopicKey: topic, + resourceSinkInputSpecsSubsetSchemaTypeKey: "", + resourceSinkInputSpecsSubsetSerdeClassNameKey: "", + resourceSinkInputSpecsSubsetIsRegexPatternKey: false, + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: defaultSinkReceiverQueueSize, + resourceSinkInputSpecsSubsetPoolMessagesKey: false, + } + for key, value := range overrides { + spec[key] = value + } + + return spec +} + +func sinkResourceData(t *testing.T, values map[string]interface{}) *schema.ResourceData { + t.Helper() + + d := schema.TestResourceDataRaw(t, resourcePulsarSink().Schema, map[string]interface{}{}) + for key, value := range values { + require.NoError(t, d.Set(key, value)) + } + + return d +} + +// The point of #218: tuning the queue size must not force the user to also name a schema type and +// a serde class, which Pulsar rejects together anyway. +func TestMarshalSinkInputSpecsQueueSizeOnly(t *testing.T) { + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkTenantKey: "public", + resourceSinkNamespaceKey: "default", + resourceSinkNameKey: "sink-1", + resourceSinkInputSpecsKey: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 100, + }), + }, + }) + + sinkConfig, err := marshalSinkConfig(d) + require.NoError(t, err) + + spec, ok := sinkConfig.InputSpecs["persistent://public/default/in-1"] + require.True(t, ok) + assert.Equal(t, 100, spec.ReceiverQueueSize) + assert.True(t, spec.HasReceiverQueueSize()) + assert.Empty(t, spec.SchemaType) + assert.Empty(t, spec.SerdeClassName) + assert.Nil(t, spec.ConsumerProperties) +} + +func TestMarshalSinkInputSpecsSupportedFields(t *testing.T) { + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkTenantKey: "public", + resourceSinkNamespaceKey: "default", + resourceSinkNameKey: "sink-1", + resourceSinkInputSpecsKey: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetPoolMessagesKey: true, + resourceSinkInputSpecsSubsetConsumerPropertiesKey: map[string]interface{}{ + "application": "billing", + }, + }), + }, + }) + + sinkConfig, err := marshalSinkConfig(d) + require.NoError(t, err) + + spec := sinkConfig.InputSpecs["persistent://public/default/in-1"] + assert.True(t, spec.PoolMessages) + assert.Equal(t, map[string]string{"application": "billing"}, spec.ConsumerProperties) +} + +func TestMarshalSinkInputSpecsExplicitZeroQueueSize(t *testing.T) { + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkTenantKey: "public", + resourceSinkNamespaceKey: "default", + resourceSinkNameKey: "sink-1", + resourceSinkInputSpecsKey: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 0, + }), + }, + }) + + sinkConfig, err := marshalSinkConfig(d) + require.NoError(t, err) + spec := sinkConfig.InputSpecs["persistent://public/default/in-1"] + assert.True(t, spec.HasReceiverQueueSize()) + assert.Zero(t, spec.ReceiverQueueSize) +} + +func TestValidateSinkInputSpecs(t *testing.T) { + tests := []struct { + name string + specs []interface{} + wantErr string + }{ + { + name: "duplicate topic keys are rejected", + specs: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 100, + }), + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 999, + }), + }, + wantErr: "duplicate", + }, + { + // SinkConfigUtils rejects this server-side; catching it at plan time is a better error. + name: "schema_type and serde_class_name are mutually exclusive", + specs: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetSchemaTypeKey: "avro", + resourceSinkInputSpecsSubsetSerdeClassNameKey: "com.acme.MySerde", + }), + }, + wantErr: "cannot set both", + }, + { + name: "queue size alone is valid", + specs: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 100, + }), + }, + }, + { + name: "distinct topics are valid", + specs: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 100, + }), + sinkInputSpec("persistent://public/default/in-2", map[string]interface{}{ + resourceSinkInputSpecsSubsetSchemaTypeKey: "avro", + }), + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkInputSpecsKey: test.specs, + }) + + err := validateSinkInputSpecs(d.Get(resourceSinkInputSpecsKey)) + if test.wantErr == "" { + assert.NoError(t, err) + return + } + require.Error(t, err) + assert.Contains(t, err.Error(), test.wantErr) + }) + } +} + +func TestMarshalSinkInputSpecsFilterLegacyOverlaps(t *testing.T) { + topics := []string{ + "persistent://public/default/plain", + "persistent://public/default/pattern-.*", + "persistent://public/default/serde", + "persistent://public/default/schema", + } + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkTenantKey: "public", + resourceSinkNamespaceKey: "default", + resourceSinkNameKey: "sink-1", + resourceSinkInputsKey: []interface{}{topics[0]}, + resourceSinkTopicsPatternKey: topics[1], + resourceSinkCustomSerdeInputsKey: map[string]interface{}{topics[2]: "com.acme.Serde"}, + resourceSinkCustomSchemaInputsKey: map[string]interface{}{ + topics[3]: "STRING", + }, + resourceSinkInputSpecsKey: []interface{}{ + sinkInputSpec(topics[0], nil), + sinkInputSpec(topics[1], map[string]interface{}{ + resourceSinkInputSpecsSubsetIsRegexPatternKey: true, + }), + sinkInputSpec(topics[2], map[string]interface{}{ + resourceSinkInputSpecsSubsetSerdeClassNameKey: "com.acme.Serde", + }), + sinkInputSpec(topics[3], map[string]interface{}{ + resourceSinkInputSpecsSubsetSchemaTypeKey: "STRING", + }), + }, + }) + + sinkConfig, err := marshalSinkConfig(d) + require.NoError(t, err) + assert.Nil(t, sinkConfig.Inputs) + assert.Nil(t, sinkConfig.TopicsPattern) + assert.Nil(t, sinkConfig.TopicToSerdeClassName) + assert.Nil(t, sinkConfig.TopicToSchemaType) + assert.Len(t, sinkConfig.InputSpecs, len(topics)) +} + +func TestSinkInputSpecsValidationDuringPlan(t *testing.T) { + res := resourcePulsarSink() + d := schema.TestResourceDataRaw(t, res.Schema, map[string]interface{}{}) + config := sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{ + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 100, + }), + sinkInputSpec("persistent://public/default/in-1", map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 200, + }), + }, + }) + + _, err := res.Diff(context.Background(), d.State(), terraform.NewResourceConfigRaw(config), nil) + require.ErrorContains(t, err, "duplicate") +} + +func TestUnmarshalSinkInputSpecsPreservesLegacyRepresentation(t *testing.T) { + topic := "persistent://public/default/in-1" + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkInputsKey: []interface{}{topic}, + }) + consumerConfig := utils.ConsumerConfig{} + consumerConfig.SetReceiverQueueSize(100) + + err := unmarshalSinkInputSpecs(utils.SinkConfig{ + InputSpecs: map[string]utils.ConsumerConfig{topic: consumerConfig}, + }, d) + require.NoError(t, err) + assert.Empty(t, d.Get(resourceSinkInputSpecsKey).(*schema.Set).List()) + assert.Equal(t, []interface{}{topic}, d.Get(resourceSinkInputsKey).(*schema.Set).List()) +} + +func TestUnmarshalSinkInputSpecsRefreshesLegacyRepresentation(t *testing.T) { + topics := map[string]string{ + "plain": "persistent://public/default/plain", + "removed": "persistent://public/default/removed", + "became_pattern": "persistent://public/default/became-pattern", + "pattern": "persistent://public/default/pattern-.*", + "serde": "persistent://public/default/serde", + "schema": "persistent://public/default/schema", + } + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkInputsKey: []interface{}{ + topics["plain"], topics["removed"], topics["became_pattern"], + }, + resourceSinkTopicsPatternKey: topics["pattern"], + resourceSinkCustomSerdeInputsKey: map[string]interface{}{ + topics["serde"]: "com.acme.OldSerde", + topics["removed"]: "com.acme.RemovedSerde", + }, + resourceSinkCustomSchemaInputsKey: map[string]interface{}{ + topics["schema"]: "STRING", + topics["removed"]: "BYTES", + }, + }) + + err := unmarshalSinkInputSpecs(utils.SinkConfig{InputSpecs: map[string]utils.ConsumerConfig{ + topics["plain"]: {}, + topics["became_pattern"]: { + RegexPattern: true, + }, + topics["pattern"]: { + RegexPattern: true, + }, + topics["serde"]: { + SerdeClassName: "com.acme.NewSerde", + }, + topics["schema"]: { + SchemaType: "AVRO", + }, + }}, d) + require.NoError(t, err) + + assert.Equal(t, []interface{}{topics["plain"]}, d.Get(resourceSinkInputsKey).(*schema.Set).List()) + assert.Equal(t, topics["pattern"], d.Get(resourceSinkTopicsPatternKey)) + assert.Equal(t, map[string]interface{}{topics["serde"]: "com.acme.NewSerde"}, + d.Get(resourceSinkCustomSerdeInputsKey)) + assert.Equal(t, map[string]interface{}{topics["schema"]: "AVRO"}, + d.Get(resourceSinkCustomSchemaInputsKey)) + + specs := d.Get(resourceSinkInputSpecsKey).(*schema.Set).List() + require.Len(t, specs, 1) + assert.Equal(t, topics["became_pattern"], + specs[0].(map[string]interface{})[resourceSinkInputSpecsSubsetTopicKey]) +} + +func TestUnmarshalImportedSinkInputsUsesLegacyFieldsWhenLossless(t *testing.T) { + topics := map[string]string{ + "plain": "persistent://public/default/plain", + "pattern": "persistent://public/default/pattern-.*", + "serde": "persistent://public/default/serde", + "schema": "persistent://public/default/schema", + "advanced": "persistent://public/default/advanced", + } + advanced := utils.ConsumerConfig{ + PoolMessages: true, + ConsumerProperties: map[string]string{"application": "billing"}, + } + advanced.SetReceiverQueueSize(0) + d := sinkResourceData(t, nil) + + err := unmarshalSinkInputSpecs(utils.SinkConfig{InputSpecs: map[string]utils.ConsumerConfig{ + topics["plain"]: {}, + topics["pattern"]: { + RegexPattern: true, + }, + topics["serde"]: { + SerdeClassName: "com.acme.Serde", + }, + topics["schema"]: { + SchemaType: "AVRO", + }, + topics["advanced"]: advanced, + }}, d) + require.NoError(t, err) + + assert.Equal(t, []interface{}{topics["plain"]}, d.Get(resourceSinkInputsKey).(*schema.Set).List()) + assert.Equal(t, topics["pattern"], d.Get(resourceSinkTopicsPatternKey)) + assert.Equal(t, map[string]interface{}{topics["serde"]: "com.acme.Serde"}, + d.Get(resourceSinkCustomSerdeInputsKey)) + assert.Equal(t, map[string]interface{}{topics["schema"]: "AVRO"}, + d.Get(resourceSinkCustomSchemaInputsKey)) + + specs := d.Get(resourceSinkInputSpecsKey).(*schema.Set).List() + require.Len(t, specs, 1) + spec := specs[0].(map[string]interface{}) + assert.Equal(t, topics["advanced"], spec[resourceSinkInputSpecsSubsetTopicKey]) + assert.Zero(t, spec[resourceSinkInputSpecsSubsetReceiverQueueSizeKey]) + assert.True(t, spec[resourceSinkInputSpecsSubsetPoolMessagesKey].(bool)) + assert.Equal(t, map[string]interface{}{"application": "billing"}, + spec[resourceSinkInputSpecsSubsetConsumerPropertiesKey]) +} + +func TestUnmarshalSinkInputSpecsPreservesExplicitZero(t *testing.T) { + topic := "persistent://public/default/in-1" + d := sinkResourceData(t, map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{ + sinkInputSpec(topic, map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 0, + }), + }, + }) + consumerConfig := utils.ConsumerConfig{} + consumerConfig.SetReceiverQueueSize(0) + + err := unmarshalSinkInputSpecs(utils.SinkConfig{ + InputSpecs: map[string]utils.ConsumerConfig{topic: consumerConfig}, + }, d) + require.NoError(t, err) + specs := d.Get(resourceSinkInputSpecsKey).(*schema.Set).List() + require.Len(t, specs, 1) + assert.Zero(t, specs[0].(map[string]interface{})[resourceSinkInputSpecsSubsetReceiverQueueSizeKey]) +} + +func sinkConfigWithBase(values map[string]interface{}) map[string]interface{} { + config := map[string]interface{}{ + resourceSinkTenantKey: "public", + resourceSinkNamespaceKey: "default", + resourceSinkNameKey: "sink-1", + resourceSinkCleanupSubscriptionKey: false, + resourceSinkArchiveKey: "builtin://jdbc", + } + for key, value := range values { + config[key] = value + } + return config +} + +func sinkInputSpecsDiff(t *testing.T, state, config map[string]interface{}) *terraform.InstanceDiff { + t.Helper() + res := resourcePulsarSink() + d := schema.TestResourceDataRaw(t, res.Schema, map[string]interface{}{}) + for key, value := range state { + require.NoError(t, d.Set(key, value)) + } + d.SetId("public/default/sink-1") + + diff, err := res.Diff(context.Background(), d.State(), terraform.NewResourceConfigRaw(config), nil) + require.NoError(t, err) + require.NotNil(t, diff) + return diff +} + +func TestSinkImportedLegacyInputPlansCleanly(t *testing.T) { + topic := "persistent://public/default/in-1" + res := resourcePulsarSink() + d := schema.TestResourceDataRaw(t, res.Schema, sinkConfigWithBase(nil)) + d.SetId("public/default/sink-1") + + require.NoError(t, unmarshalSinkInputSpecs(utils.SinkConfig{ + InputSpecs: map[string]utils.ConsumerConfig{topic: {}}, + }, d)) + + config := sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputsKey: []interface{}{topic}, + }) + diff, err := res.Diff(context.Background(), d.State(), terraform.NewResourceConfigRaw(config), nil) + require.NoError(t, err) + if diff != nil { + assert.True(t, diff.Empty(), "unexpected import follow-up diff: %#v", diff.Attributes) + } +} + +func TestSinkExistingDefaultQueueStatePlansCleanly(t *testing.T) { + topic := "persistent://public/default/in-1" + res := resourcePulsarSink() + d := schema.TestResourceDataRaw(t, res.Schema, map[string]interface{}{}) + for key, value := range sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{sinkInputSpec(topic, nil)}, + }) { + require.NoError(t, d.Set(key, value)) + } + d.SetId("public/default/sink-1") + + config := sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{ + map[string]interface{}{resourceSinkInputSpecsSubsetTopicKey: topic}, + }, + }) + diff, err := res.Diff(context.Background(), d.State(), terraform.NewResourceConfigRaw(config), nil) + require.NoError(t, err) + if diff != nil { + assert.True(t, diff.Empty(), "existing queue state re-planned: %#v", diff.Attributes) + } +} + +func TestSinkInputSpecsForceNew(t *testing.T) { + topic := "persistent://public/default/in-1" + tests := []struct { + name string + state map[string]interface{} + config map[string]interface{} + requiresNew bool + }{ + { + name: "queue size updates in place", + state: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{sinkInputSpec(topic, map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 100, + })}, + }), + config: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{sinkInputSpec(topic, map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 250, + })}, + }), + }, + { + name: "adopting input_specs for an existing topic updates in place", + state: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputsKey: []interface{}{topic}, + }), + config: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputsKey: []interface{}{topic}, + resourceSinkInputSpecsKey: []interface{}{sinkInputSpec(topic, map[string]interface{}{ + resourceSinkInputSpecsSubsetReceiverQueueSizeKey: 250, + })}, + }), + }, + { + name: "renaming an input_specs topic replaces the sink", + state: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{sinkInputSpec(topic, nil)}, + }), + config: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{ + sinkInputSpec("persistent://public/default/renamed", nil), + }, + }), + requiresNew: true, + }, + { + name: "flipping the regex flag replaces the sink", + state: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{sinkInputSpec(topic, nil)}, + }), + config: sinkConfigWithBase(map[string]interface{}{ + resourceSinkInputSpecsKey: []interface{}{sinkInputSpec(topic, map[string]interface{}{ + resourceSinkInputSpecsSubsetIsRegexPatternKey: true, + })}, + }), + requiresNew: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + diff := sinkInputSpecsDiff(t, test.state, test.config) + assert.Equal(t, test.requiresNew, diff.RequiresNew()) + }) + } +} diff --git a/pulsar/testdata/sink/main.tf b/pulsar/testdata/sink/main.tf index cec8ebf1..5172c5fe 100644 --- a/pulsar/testdata/sink/main.tf +++ b/pulsar/testdata/sink/main.tf @@ -22,10 +22,20 @@ provider "pulsar" { resource "pulsar_sink" "sink-1" { provider = pulsar - name = "sink-1" - tenant = "public" - namespace = "default" - inputs = ["sink-1-topic"] + name = "sink-1" + tenant = "public" + namespace = "default" + inputs = ["sink-1-topic"] + # Deliberately overlap inputs and input_specs. The provider must send the complete input_specs + # entry as the canonical representation so Pulsar's update path cannot overwrite its settings. + input_specs { + key = "sink-1-topic" + receiver_queue_size = 0 + pool_messages = true + consumer_properties = { + application = "billing" + } + } subscription_position = "Latest" cleanup_subscription = false parallelism = 1