fix(sink): make input_specs attributes optional and complete

[david-streamlio]

Fixes #218

Motivation

Every nested attribute of pulsar_sink's input_specs block is Required:

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},

So tuning receiver_queue_size on one topic forces the user to also name a schema_type and a serde_class_name. That is worse than inconvenient: SinkConfigUtils.doJavaChecks() rejects a spec that sets both ("Only one of schemaType or serdeClassName should be set in inputSpec"), so the schema requires a combination the broker refuses.

utils.ConsumerConfig also carries PoolMessages, SchemaProperties and ConsumerProperties, none of which had an attribute.

Modifications

• Make every nested attribute Optional except key, which is the InputSpecs map key and so genuinely required.
• Add pool_messages, schema_properties and consumer_properties.
• Add validateSinkInputSpecs for the two rules the schema cannot express, mirroring the approach taken for pulsar_function's input_specs in feat(function): expose input_specs for per-topic consumer config #219:
  • Topic uniqueness. TypeSet hashes the whole nested object, so two blocks with the same key but different queue sizes are both valid set elements and then collapse to whichever the map iteration wrote last. Rejecting this at plan time avoids an ambiguous request and perpetual drift.
  • schema_type / serde_class_name exclusivity, matching the server, so the error arrives at plan time rather than apply time.
• Add a receiver_queue_size >= 0 validation and mark it Computed, since the broker supplies a default.
• Empty schema_properties / consumer_properties maps are omitted from the request and only written into state when the broker returns something, so an empty map does not read as configuration the user never wrote.

Backward compatibility — worth your read. Relaxing Required to Optional is not breaking for existing configurations: a config that sets all five attributes keeps working unchanged. It does mean state written by an older provider version carries explicit zero values ("", false, 0) where a user might now omit the attribute. I did not add a state upgrader, on the grounds that those zero values are what the schema previously forced and remain valid. Say the word if you would rather have one.

The attribute names are left as they are — key, serde_class_name, is_regex_pattern — rather than aligned with pulsar_function's newer input_specs block, since renaming them would be genuinely breaking.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Unit tests in resource_pulsar_sink_unit_test.go: a spec that sets only receiver_queue_size marshals correctly with no schema or serde (the case the issue is about); the three new fields round-trip; and the validation rejects duplicate topics and the schema/serde combination while accepting queue-size-only and distinct-topic specs.
• TestSink and TestSinkUpdate pass against Pulsar 4.0.3, confirming existing sink configurations are unaffected.
• make test, golangci-lint and go generate ./... are all clean.

Documentation

Check the box below.

Need to update docs?

• doc-required

• no-need-doc

• doc

  docs/resources/sink.md regenerated; the nested block now shows one required and seven optional attributes, each with a description.

[david-streamlio]

Pushed a fix for the acceptance test failure.

TestImportExistingSink asserts the exact number of flattened state attributes on an imported sink. The three attributes this PR adds to input_specs — pool_messages, schema_properties and consumer_properties — take that count from 30 to 33, so the assertion had to move with it. I have pulled the literal into a named constant and documented what it counts, since a bare number gives no indication why it changes when the schema does.

My original verification was too narrow: I ran -run 'TestSink', which matches TestSink and TestSinkUpdate but not TestImportExistingSink, so I never executed the test this broke. Re-verified with -run 'Sink' at -count 3, matching what CI does — all sink tests pass across three consecutive runs.

One thing worth flagging for the record: during that investigation I saw TestSink fail once when run alongside the other sink tests, and pass when run alone. I could not reproduce it afterwards — the same set passes on unmodified master and passes on this branch across three runs — so it looks like an existing flake rather than anything this PR introduces. Noting it in case it resurfaces here or elsewhere.

Also unrelated to this change, but visible in the failed run: the workflow's tmate debug step keeps the job alive for ~15 minutes after a failure and fills the log tail with SSH banners, which buries the actual assertion. Worth a separate look if others hit the same confusion.

[freeznet]

Pushed caf2c36 with the sink input-spec fixes found during the broker-level review.

Changes:

• preserve an explicit receiver_queue_size = 0 with the client presence API, while using the Pulsar default of 1000 only when omitted;
• move duplicate-topic and schema/serde validation into CustomizeDiff, so invalid configurations fail during planning;
• normalize overlaps across inputs, topics_pattern, custom serde/schema maps, and input_specs, preventing Pulsar's update merge from overwriting explicit consumer settings;
• apply topology-aware replacement only when the effective topic set or regex flag changes;
• preserve the configured legacy representation on refresh and use input_specs as the canonical import representation;
• retain the broker-supported pool_messages and consumer_properties;
• remove/defer schema_properties, because Pulsar 4.0.3 SinkConfigUtils drops it in both request conversion and read-back.

Regression coverage now includes explicit zero presence, plan-time validation, legacy overlap filtering, in-place consumer updates, topology replacement, import normalization, and real broker read-back.

Validation:

• GOTOOLCHAIN=go1.25.14+auto GOWORK=off go test ./...
• GOTOOLCHAIN=go1.25.14+auto GOWORK=off go vet ./...
• go generate ./...
• Pulsar 4.0.3: TestSink, TestSinkUpdate, and TestImportExistingSink all pass; the update changes queue size from 0 to 100 without replacement and retains consumer properties and pooling.

[david-streamlio]

Thanks for picking this up, @freeznet — the overlap dedup and the receiver_queue_size = 0 presence handling are real fixes, and asserting the resource ID is unchanged in TestSinkUpdate is exactly the right way to prove the ForceNew rework avoids replacement.

I went through caf2c36 in detail. Build, go vet, the unit tests and Provider().InternalValidate() all pass locally; I have not run the acceptance tests (no live cluster here).

One item I would like to resolve before this merges, plus two behaviour changes worth making explicit:

1. schema_properties was removed. It is one of the three attributes this PR exists to add, and ConsumerConfig.SchemaProperties is still live upstream. Details inline — this is the only one I would call blocking.
2. Refresh no longer writes back inputs / topics_pattern / custom_serde_inputs / custom_schema_inputs, so drift on those is no longer detected, and imported state carries no inputs at all.
3. receiver_queue_size is now always transmitted rather than omitted when unset.

Happy to push fixes for any of these myself if you would rather not carry them — just say which.

[david-streamlio]

@maxsxu ready for a re-review when you have a moment — flagging explicitly because the review request no longer shows on this PR: both your approval and @freeznet's were dismissed by the two pushes in 2754166 and c9b978c, and the cloud-data-plane request was cleared along with them, so this won't appear in your review queue.

State as of now:

• All seven review threads are resolved. Rui answered each one and the fixes are in 2754166 — I verified them against the head rather than taking the replies at face value.
• The one item I had flagged as blocking, the removal of schema_properties, is settled: the sink REST conversion path drops the field in both directions, so exposing it here would be write-only with permanent read-back drift. Tracked upstream as [Bug] Sink inputSpecs schemaProperties is silently dropped by SinkConfigUtils in both conversion directions apache/pulsar#26425; the attribute can come back once that lands.
• golangci-lint and GoReleaser check pass. The two acceptance jobs are still running against c9b978c — worth waiting for those, since this is the first acceptance run against the refresh/import rewrite.

Nothing outstanding from my side.