pulsaradmin: ProducerConfig cannot express PIP-401 batchingConfig

[david-streamlio]

Is your feature request related to a problem? Please describe.

pulsaradmin's utils.ProducerConfig cannot express PIP-401's producer batching configuration, so no Go-based tool can set it on a Pulsar Function, Source, or Sink.

PIP-401 (apache/pulsar#23860, merged 2025-03-12) added a BatchingConfig to the Java admin API's ProducerConfig and wired it through to the ProducerBuilder. The Java class today is:

// pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java
public class ProducerConfig {
    private Integer maxPendingMessages;
    private Integer maxPendingMessagesAcrossPartitions;
    private Boolean useThreadLocalProducers;
    private CryptoConfig cryptoConfig;
    private String batchBuilder;
    private CompressionType compressionType;
    private BatchingConfig batchingConfig;   // <-- added by PIP-401
}

pulsaradmin/pkg/utils/producer_config.go on master has every field except the last one:

type ProducerConfig struct {
	MaxPendingMessages int `json:"maxPendingMessages" yaml:"maxPendingMessages"`
	//nolint
	MaxPendingMessagesAcrossPartitions int `json:"maxPendingMessagesAcrossPartitions" yaml:"maxPendingMessagesAcrossPartitions"`

	UseThreadLocalProducers bool          `json:"useThreadLocalProducers" yaml:"useThreadLocalProducers"`
	CryptoConfig            *CryptoConfig `json:"cryptoConfig" yaml:"cryptoConfig"`
	BatchBuilder            string        `json:"batchBuilder" yaml:"batchBuilder"`
	CompressionType         string        `json:"compressionType" yaml:"compressionType"`
}

Since ProducerConfig is a typed struct serialized to JSON with no passthrough field, there is no workaround: a Go caller simply cannot put batchingConfig on the wire, even though the broker accepts it and the Java pulsar-admin CLI sends it happily via functions create --producer-config '<json>'.

The concrete impact is on infrastructure-as-code. The Terraform provider (streamnative/terraform-provider-pulsar) imports this package, so producer batching is unreachable from Terraform for every function runtime — including Java, where the broker and runtime support have been complete since PIP-401. pulsarctl has the same ceiling.

The user-visible symptom this traces back to is publish latency: BatchingConfig.batchingMaxPublishDelayMs defaults to 10ms, and below roughly one message per 10ms per instance every message pays that full linger while each batch still contains exactly one message. Tuning or disabling it is exactly what batchingConfig exists for.

Describe the solution you'd like

Add a BatchingConfig type mirroring the Java class and reference it from ProducerConfig:

// pulsaradmin/pkg/utils/batching_config.go
type BatchingConfig struct {
	Enabled                   bool `json:"enabled" yaml:"enabled"`
	BatchingMaxPublishDelayMs *int `json:"batchingMaxPublishDelayMs,omitempty" yaml:"batchingMaxPublishDelayMs"`
	//nolint
	RoundRobinRouterBatchingPartitionSwitchFrequency *int `json:"roundRobinRouterBatchingPartitionSwitchFrequency,omitempty" yaml:"roundRobinRouterBatchingPartitionSwitchFrequency"`
	BatchingMaxMessages *int   `json:"batchingMaxMessages,omitempty" yaml:"batchingMaxMessages"`
	BatchingMaxBytes    *int   `json:"batchingMaxBytes,omitempty" yaml:"batchingMaxBytes"`
	BatchBuilder        string `json:"batchBuilder,omitempty" yaml:"batchBuilder"`
}

and in ProducerConfig:

	BatchingConfig *BatchingConfig `json:"batchingConfig,omitempty" yaml:"batchingConfig"`

Two details worth a maintainer's eye:

1. Pointer-typed numerics. The Java fields are boxed (Integer), and BatchingConfig has non-zero defaults — enabled = true and batchingMaxPublishDelayMs = 10. Plain int with omitempty would make "explicitly 0" indistinguishable from "unset", which matters most for batchingMaxPublishDelayMs, where 0 is the meaningful value for latency-sensitive callers who want the linger gone. *int keeps that distinction. Enabled is a plain bool because Java declares it unboxed with a true default; a caller wanting to disable batching sets false explicitly, which must serialize — hence no omitempty on it.
2. ProducerConfig as a pointer field. BatchingConfig should be *BatchingConfig so an unset batching config is omitted entirely rather than sent as a zero-valued object, which would read to the broker as enabled: false.

ProducerConfig is already referenced by FunctionConfig, SourceConfig, and SinkConfig, so a single addition covers all three resource types.

Describe alternatives you've considered

• A free-form passthrough field on ProducerConfig — rejected. It would diverge from the Java model and push JSON assembly onto every caller.
• Doing this in the Terraform provider instead — not possible. The provider serializes through this struct; there is no seam to inject an extra JSON key.
• customRuntimeOptions — not a workaround. It maps to a different FunctionDetails field that the worker passes to the runtime, not into producerSpec.

Additional context

Runtime support for batchingSpec varies, which affects whether setting this has an effect but not whether it can be set:

• Java — complete since PIP-401.
• Python — ignores it today ([Functions] Python function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec pulsar#26390). A fix is proposed in [fix][fn] Honour producerSpec batching configuration in the Python function runtime pulsar#26392, open and not merged.
• Go — ignores it today ([Functions] Go function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec pulsar#26391). A fix is proposed in [fix][fn] Honour producerSpec batching configuration in the Go function runtime pulsar#26393, open and not merged. Note this is the pulsar-function-go runtime in the apache/pulsar repo, a separate concern from this admin-client change.

To be explicit, since it would be easy to read the above as more settled than it is: neither runtime PR has been reviewed or merged, and this request does not depend on either landing. The gap described here applies today to Java functions, where the runtime has honoured batchingSpec since PIP-401 merged in March 2025 — a Go caller still cannot send the configuration that runtime already supports. The Python and Go runtime work would widen who benefits; it is not a prerequisite.

Downstream, streamnative/terraform-provider-pulsar#220 tracks exposing producerConfig on pulsar_function and is blocked on this issue for the batching portion.

Happy to open a PR for this if the shape above looks right — in particular I would like a maintainer's read on the pointer-vs-value choice before writing it.

[david-streamlio]

@freeznet has implemented this in #1529 — thank you. Linking the two so this issue reflects that.

The implementation there is a better shape than the *int I sketched above: ReceiverQueueSize int tagged json:"-" alongside an unexported receiverQueueSizeSet bool, with MarshalJSON/UnmarshalJSON mapping it to a *int on the wire and a SetReceiverQueueSize() / HasReceiverQueueSize() pair. That keeps the exported field an int, so existing callers assigning a non-zero value compile and behave exactly as before, while an explicit zero still serializes. My proposal would have changed the field's type and broken those callers — this avoids that entirely.

Verified end to end against Pulsar 4.0.3 while reviewing the downstream Terraform change: a function configured with receiverQueueSize: 0 round-trips with HasReceiverQueueSize() == true and value 0, where previously omitempty dropped it and the broker applied the 1000 default.

Note this covers the receiverQueueSize half of what I filed. The batchingConfig half — PIP-401's BatchingConfig still being absent from utils.ProducerConfig — is untouched by #1529 and remains open. Happy to split that into its own issue if that would make tracking cleaner once #1529 merges.

[david-streamlio]

Picking up the remaining half of this — adding PIP-401's BatchingConfig to utils.ProducerConfig — so it does not get duplicated. #1529 covers receiverQueueSize and does not touch ProducerConfig; there is no other open PR adding BatchingConfig.

I will follow the idiom #1529 established rather than the *int shape I sketched in the description above, so the two land consistently: keep exported fields as plain value types and track explicitness separately where a zero value is meaningful, rather than changing field types in a way that breaks existing callers.

ProducerConfig is shared by FunctionConfig, SourceConfig and SinkConfig, so this covers all three resource types in one change. I will open a PR against master and link it here.

@freeznet — flagging since you are working in this area: shout if you have already started on this and I will drop it.

[david-streamlio]

Opened #1530 for this half.

One correction to the description above, found while verifying against a real broker: batchingConfig was added to ProducerConfig in Apache Pulsar 4.1.0, not in the 4.0.x line. BatchingConfig.java returns 404 on tags v4.0.3 through v4.0.6 and first appears in v4.1.0. The Java class I quoted was from master. #1530 therefore omits the field when nil so requests to pre-4.1.0 brokers are byte-identical to before.

A second thing worth recording, because it makes this field behave the opposite way to #1529: batchingMaxPublishDelayMs: 0 does not remove the batching linger. BatchingUtils.convert() applies each numeric field only when it is present and greater than zero, so zero is read as unconfigured. Confirmed against 4.1.0 — a function created with zero reads back as 10. Batching is turned off with enabled: false, not with a zero delay. So unlike receiverQueueSize, there is no unset-versus-zero problem to solve here, which is why #1530 uses plain nullable pointers rather than the setter idiom from #1529.