feat(function): expose input_specs for per-topic consumer config

[david-streamlio]

Fixes #217

Motivation

pulsar_function supports inputs, custom_serde_inputs, custom_schema_inputs and friends, but there is no way to set Pulsar's inputSpecs — the per-topic consumer configuration. The practical consequence is that receiverQueueSize is unreachable from Terraform. It defaults to 1000, so a Shared-subscription function at high parallelism buffers up to that many messages per instance (at 15 instances, up to 15,000 messages parked in memory). The only way to change it today is pulsarctl, i.e. outside the user's Terraform, where the next apply can neither see nor preserve the change.

utils.FunctionConfig.InputSpecs and utils.ConsumerConfig are already present in the admin client the provider depends on — only the schema and its marshal/unmarshal were missing.

Modifications

Adds an input_specs block to pulsar_function, deliberately mirroring the block pulsar_sink already ships so the two resources read the same: key, schema_type, serde_class_name, is_regex_pattern, receiver_queue_size, plus pool_messages, schema_properties and consumer_properties. Unlike the sink block, only key is Required (the sink's all-Required block is #218, not touched here). crypto_config is intentionally out of scope.

Three behaviours in FunctionConfigUtils.java shape the implementation, and each is worth a reviewer's attention:

1. The read path never returns inputs. convertFromDetails() builds a ConsumerConfig for every input topic — including topics declared through inputs, and the topics_pattern entry as a spec with isRegexPattern=true — and only calls setInputSpecs(). Mirroring the response into state would invent input_specs blocks for configurations that never wrote one, producing a diff on every plan forever. So a returned spec is skipped when inputs or topics_pattern already represents that topic, unless the configuration also declares it in input_specs (in which case it is genuinely the user's and is refreshed rather than dropped).

2. inputs clobbers inputSpecs on update. In validateUpdate(), each entry of newConfig.getInputs() is written into newConfig.getInputSpecs() with a fresh ConsumerConfig before that map is iterated, so a topic listed in both would silently lose its receiverQueueSize on every apply. (convert() applies inputSpecs last on create, so the two paths disagree.) Topics carried by an input_specs block are therefore stripped from inputs on the wire.

3. The topic set and isRegexPattern are immutable, everything else is not. validateUpdate() throws "Input Topics cannot be altered" and rejects isRegexPattern changes, but then does mergedConfig.getInputSpecs().put(topicName, consumerConfig) — so receiver_queue_size genuinely updates in place. CustomizeDiff mirrors exactly that rule, comparing the effective topic set (the union of inputs, topics_pattern and input_specs). This matters for adoption: moving a topic that inputs already declares into an input_specs block leaves that set untouched and stays an in-place update, rather than destroying the user's function.

Two SDK details are called out in comments because they are easy to reintroduce:

• Nested ForceNew is deliberately absent. On a TypeSet, changing receiver_queue_size rehashes the element, which the SDK reads as a removal plus an addition — a nested ForceNew would replace the function on the very edit this feature exists to enable.
• A set-level diff.ForceNew() is not sufficient on its own. schemaMap.diffSet only carries it into the diff through the input_specs.# count attribute, which it emits solely when the element count changes; renaming a topic or flipping is_regex_pattern keeps the count identical and the replacement would be silently dropped. The changed nested attribute is flagged as well.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Unit tests (no cluster needed) covering marshal (including the inputs overlap strip and empty-map omission), unmarshal (the inputs-covered skip, the both-places case, and the import case), and the effective-topic-set computation.
• TestFunctionInputSpecsForceNew — a table-driven test running the real schema.Resource.Diff over six scenarios: adopting input_specs for a topic already in inputs and tuning receiver_queue_size must not replace; adding, renaming, dropping a topic and flipping is_regex_pattern must. The two in-place cases additionally assert the planned value actually reaches the diff, so a change the SDK dropped entirely cannot pass as "no replacement needed".
• TestFunction (acceptance) — extended with a second step raising receiver_queue_size from 100 to 250, asserting the broker reports the new value, that schema_type survived (which fails if the topic was left in inputs, per finding 2), and that the resource ID is unchanged so the update was in place rather than a recreate. Run against Pulsar in Docker, -count 3.
• TestFunctionLegacyInputsOnly (acceptance) — a regression guard that a function configured only with inputs does not drift, since resource.Test fails a step whose post-apply plan is non-empty. This is the main risk to existing users, per finding 1.

One side effect worth noting: import fidelity improves. inputs was already never populated on import because the API does not return it, so imported functions now capture their input topics as input_specs blocks instead of losing them entirely.

Documentation

Check the box below.

Need to update docs?

• doc-required

• no-need-doc

• doc

  docs/resources/function.md regenerated with go generate ./...; the diff there is purely additive.

[freeznet]

Blocking correctness issues remain against the repository's Pulsar 4.0.3 acceptance baseline and the pinned pulsar-client-go model. The core approach and the tests for plain inputs overlap and in-place queue-size updates are good, but the other legacy input forms and several newly exposed values do not yet round-trip safely. Please address the inline findings and add focused regression or acceptance coverage.

[freeznet]

apache/pulsar-client-go#1529 need this fix, once it get merged, will change the go mod to upstream master

[david-streamlio]

@freeznet quick question on the replace directive, and some context on why I'm asking.

This is driving a customer PoC where Terraform is a hard requirement rather than a convenience: they manage Pulsar entirely as code, so pulsarctl is not an option for them. The function is Python today, Go longer term. All they need from this PR is the ability to tune receiver_queue_size down from the 1000 default to a non-zero value — the explicit-zero case does not come into it for them.

This PR is approved and green, and the only thing holding it is the replace pointing at your fork while apache/pulsar-client-go#1529 is in review. #1529 has no reviewer assigned yet, so I cannot tell whether we are waiting days or longer.

Would you be open to landing this without the explicit-zero support, and adding it back once #1529 is released upstream? The coupling is small — two call sites (SetReceiverQueueSize at resource_pulsar_function.go:852 and HasReceiverQueueSize at :1275), plus the zero-queue acceptance step and its testdata/function/main_zero_queue.tf. I would drop those, remove the replace and the go.mod/go.sum changes, and open a follow-up to restore zero support against a tagged release. Happy to do that work and re-run the acceptance suite against 4.0.3.

Entirely your call, and I am not trying to rush your fix — if you think #1529 will be reviewed shortly then waiting is cleaner and I would rather not churn your work for a few days. Mostly I want to avoid a provider release depending on a personal fork, and decoupling the two timelines seemed like the way to do that.

Either way, thanks for the review here — the inputs / customSerdeInputs clobbering and the legacy-drift-to-replacement issue were both real, and I had missed them.

[david-streamlio]

@freeznet — since you reviewed this one, a short summary of what else is now open here, so it is one notification rather than six. All are green and mergeable; none is urgent.

Blocked on upstream, not on review:

• feat(function): expose input_specs for per-topic consumer config #219 (this PR) — approved, waiting on fix(pulsaradmin): preserve explicit zero receiver queue size apache/pulsar-client-go#1529 so the replace directive can come out. My question above about landing it without the explicit-zero support still stands if #1529 is not close.

Ready to review:

• docs: warn that connector configs are stored in plaintext, and document secrets provider behaviour #204 — docs only. secrets has been supported since v0.2.0 but nothing said why it exists; adds the plaintext warning and the ClearTextSecretsProvider caveat to the schema descriptions. Rewritten from scratch after the README restructure made the original obsolete.
• fix: mark connector and function user config as sensitive #221 — Sensitive: true on configs and user_config. The enforcement half of docs: warn that connector configs are stored in plaintext, and document secrets provider behaviour #204. Names a trade-off in the body I would value your read on: non-secret values in those attributes now render as (sensitive value).
• docs: add working pulsar_schema examples, including JSON #222 — docs only, adds working pulsar_schema examples including the JSON case that Documentation error for pulsar_schema (Resource) #165 reported as wrong. Literals copied from the passing acceptance tests.
• feat(function): expose producer configuration on pulsar_function #223 — producerConfig on pulsar_function, mirroring the attribute names pulsar_source already uses. Part A of pulsar_function: producerConfig not exposed (compression_type, batch_builder, max_pending_messages), unlike pulsar_source #220 only; the batching_config block needs feat(pulsaradmin): add PIP-401 batchingConfig to ProducerConfig apache/pulsar-client-go#1530 first.
• fix(sink): make input_specs attributes optional and complete #224 — sink input_specs attributes relaxed from Required to Optional plus the three missing ConsumerConfig fields. The Required→Optional change is the part I would most like a maintainer's opinion on; I argued in the body that no state upgrader is needed, but that is a judgement call.

If you only have time for one, #221 is the smallest and the most security-relevant.