pulsar_function: producerConfig not exposed (compression_type, batch_builder, max_pending_messages), unlike pulsar_source

[david-streamlio]

Community Note

• Please vote on this issue by adding a
  👍 reaction to the original
  issue to help the community and maintainers prioritize this request
• Please do not leave "+1" or other comments that do not add relevant new information or questions, they generate extra
  noise for issue followers and do not help prioritize the request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Description

pulsar_source exposes the source's producerConfig as top-level attributes - max_pending_messages, max_pending_messages_across_partitions, use_thread_local_producers, batch_builder, compression_type, and the producer crypto settings (resource_pulsar_source.go:55-65).

pulsar_function exposes none of it. utils.FunctionConfig carries the same ProducerConfig *ProducerConfig field (plus a top-level BatchBuilder), but marshalFunctionConfig never populates it, so it is always sent nil. A function's output producer is therefore stuck on broker/runtime defaults, and no amount of Terraform configuration can change its compression, batch builder, or pending-message queue sizing.

This is the same asymmetry that #217 covers on the consumer side, in the opposite direction: there the function was missing a block the sink already had; here it is missing a surface the source already has.

Suggested scope, matching the source's attribute names so the two resources read the same:

attribute	utils.ProducerConfig field
max_pending_messages	MaxPendingMessages
max_pending_messages_across_partitions	MaxPendingMessagesAcrossPartitions
use_thread_local_producers	UseThreadLocalProducers
batch_builder	BatchBuilder
compression_type	CompressionType

The producer crypto attributes could follow the source too, though they are less urgent and could reasonably be split out.

Batching is a follow-on, and is blocked upstream

The motivating case is publish latency. Pulsar Functions got a real producer-batching configuration path in PIP-401 (apache/pulsar#23860), which added BatchingConfig to ProducerConfig and wired it through to the ProducerBuilder for the Java runtime. Two open Pulsar issues track the runtimes that still ignore it:

• [Functions] Python function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec apache/pulsar#26390 - the Python function runtime hardcodes batching_max_publish_delay_ms=10 and drops batchingSpec. Fix proposed in [fix][fn] Honour producerSpec batching configuration in the Python function runtime apache/pulsar#26392, open and not merged.
• [Functions] Go function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec apache/pulsar#26391 - the Go function runtime hardcodes BatchingMaxPublishDelay: time.Millisecond * 10 and drops batchingSpec. Fix proposed in [fix][fn] Honour producerSpec batching configuration in the Go function runtime apache/pulsar#26393, open and not merged.

Those are runtime bugs and do not require any change here - they are about honouring what already arrives in the FunctionDetails protobuf, and neither has been reviewed or merged. But note the ordering if the goal is "batching is tunable from IaC":

1. utils.ProducerConfig in pulsar-client-go's pulsaradmin has no BatchingConfig field - it carries only MaxPendingMessages, MaxPendingMessagesAcrossPartitions, UseThreadLocalProducers, CryptoConfig, BatchBuilder and CompressionType. PIP-401's addition is not modelled there, so there is currently nothing for provider code to serialize into. That needs an upstream pulsar-client-go change first, now tracked as pulsaradmin: ProducerConfig cannot express PIP-401 batchingConfig apache/pulsar-client-go#1528.
2. Then a batching_config block could be added here. At that point batching would be Terraform-configurable for Java functions, whose runtime has honoured batchingSpec since PIP-401.
3. Python and Go functions would additionally need [fix][fn] Honour producerSpec batching configuration in the Python function runtime apache/pulsar#26392 / #26393 to merge before the setting took effect for them.

Everything in the table above is reachable today with no upstream dependency, which is why this issue is scoped to producerConfig generally rather than to batching alone.

Note that custom_runtime_options is not a workaround - it maps to customRuntimeOptions, which the worker passes to the runtime, not into producerSpec.

New or Affected Resource(s)

• pulsar_function

Potential Terraform Configuration

resource "pulsar_function" "function-1" {
  provider = pulsar

  tenant    = "public"
  namespace = "default"
  name      = "function-1"

  jar       = "function://public/default/api-examples@v1"
  classname = "org.apache.pulsar.functions.api.examples.WordCountFunction"

  inputs = ["public/default/input1"]
  output = "public/default/test-out"

  # None of these are settable today.
  compression_type                       = "ZSTD"
  batch_builder                          = "KEY_BASED"
  max_pending_messages                   = 1000
  max_pending_messages_across_partitions = 50000
  use_thread_local_producers             = false
}

References

• pulsar_function: inputSpecs / receiverQueueSize not exposed, making per-topic consumer config unreachable from IaC #217 - the mirror-image gap on the consumer side (input_specs)
• [improve][fn] Implement pip 401: Support set batching configurations for Pulsar Functions&Sources apache/pulsar#23860 - PIP-401, which added BatchingConfig to ProducerConfig
• [Functions] Python function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec apache/pulsar#26390, [Functions] Go function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec apache/pulsar#26391 - Python and Go runtimes ignoring batchingSpec

[david-streamlio]

The upstream dependency mentioned above now has an issue tracking it: apache/pulsar-client-go#1528 — pulsaradmin's utils.ProducerConfig cannot express PIP-401's batchingConfig. I confirmed the field is still absent on pulsar-client-go master, not just in the version pinned here.

That splits this issue cleanly into two independently shippable halves:

Buildable today, no upstream dependency — compression_type, batch_builder, max_pending_messages, max_pending_messages_across_partitions, use_thread_local_producers. All already present on utils.ProducerConfig, and all mirroring attribute names pulsar_source already uses.

Blocked on apache/pulsar-client-go#1528 — a batching_config block. ProducerConfig is a typed struct serialized to JSON with no passthrough field, so there is no seam for the provider to inject batchingConfig itself; it has to land in the admin client first.

I would suggest shipping the first half without waiting.

For anyone arriving from the batching angle, the runtime side is tracked separately and does not gate this issue — apache/pulsar#26390 (Python) and apache/pulsar#26391 (Go). Fixes are proposed in apache/pulsar#26392 and apache/pulsar#26393 respectively; both are open and not merged. Those determine whether the setting takes effect once sent, not whether it can be sent. Java has honoured it since PIP-401, which means the admin-client fix plus the first half above would make batching Terraform-configurable for Java functions immediately.

[david-streamlio]

Two corrections to what I wrote above, both found while implementing the upstream client change.

1. batchingConfig requires Pulsar 4.1.0 or later. I said batching would become Terraform-configurable for Java functions once the client and provider changes landed, on the basis that the Java runtime has honoured batchingSpec since PIP-401. That is only true against a 4.1.0+ broker. BatchingConfig.java returns 404 on tags v4.0.3 through v4.0.6 and first appears in v4.1.0; ProducerConfig on v4.0.3 has no batchingConfig field at all.

This matters here specifically because this repository's acceptance tests run against Pulsar 4.0.3 (hack/pulsarimage/Dockerfile). A batching_config block could not be acceptance-tested on the current baseline — it would need the test image bumped, or the tests scoped to skip it. Worth settling before that half is picked up.

2. Setting the batching delay to zero does not remove the latency floor. BatchingUtils.convert() applies each numeric field only when it is present and greater than zero, so zero is treated as unconfigured. Verified against a real 4.1.0 broker: a function created with batchingMaxPublishDelayMs: 0 reads back as 10. Whatever the eventual Terraform attribute looks like, disabling batching has to go through enabled = false, and the docs should say so — a user setting the delay to zero expecting no linger would silently get the 10ms default.

Neither correction affects the first half of this issue. compression_type, batch_builder, max_pending_messages, max_pending_messages_across_partitions and use_thread_local_producers are all on utils.ProducerConfig today, work against 4.0.3, and remain buildable with no upstream dependency.

The upstream client change for the batching half is now open as apache/pulsar-client-go#1530.