feat(pulsaradmin): add PIP-401 batchingConfig to ProducerConfig

[david-streamlio]

Master Issue: #1528

Motivation

utils.ProducerConfig cannot express producer batching, so no Go caller can set it on a Function, Source or Sink — even though the broker accepts it and the Java admin API has carried it since PIP-401 (apache/pulsar#23860).

ProducerConfig is a typed struct serialized to JSON with no passthrough field, so there is no workaround: the key simply cannot be put on the wire from Go. pulsar-admin functions create --producer-config '{"batchingConfig":{...}}' works today; the equivalent is impossible from pulsarctl or the Terraform provider.

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

This is the remaining half of #1528; #1529 covered receiverQueueSize and does not touch ProducerConfig.

Modifications

Add a BatchingConfig type mirroring the Java model and reference it from ProducerConfig. Since ProducerConfig is shared by FunctionConfig, SourceConfig and SinkConfig, this covers all three resource types in one change.

Two decisions worth a reviewer's attention, both driven by how the broker consumes the payload rather than by Go style:

Enabled is serialized unconditionally — no omitempty. BatchingUtils.convert() does:

BatchingSpec batchingSpec = new BatchingSpec().setEnabled(config.isEnabled());

That reads a primitive boolean with no null fallback, so a payload that omits enabled risks arriving as false and disabling batching rather than defaulting to on. Emitting it always makes the caller's intent explicit on the wire.

The corollary is that Go's zero value is a hazard: &BatchingConfig{BatchingMaxMessages: ptr(100)} would serialize enabled: false and disable batching while appearing to cap batch size. NewBatchingConfig() returns the same defaults the broker applies when no configuration is present (enabled: true, batchingMaxPublishDelayMs: 10, matching BatchingUtils.convertFromSpec(null)), so the common "tune one field" path is safe. This follows the existing NewAutoSubscriptionCreationOverride() precedent in this package.

The numeric fields are *int, mirroring the boxed Integer fields in the Java model, so "not configured" stays distinguishable from a configured value.

I deliberately did not reuse the SetX() / HasX() + custom-marshaller idiom from #1529. That idiom exists there to avoid changing an existing exported field's type and breaking callers; BatchingConfig is a new type with no such constraint, so plain nullable pointers map the Java semantics directly with less machinery. Happy to switch to the #1529 idiom for consistency if you would rather the two match.

Note that the broker applies each numeric field only when it is present and greater than zero, so zero is not a way to switch a limit off — unlike receiverQueueSize in #1529, where zero is meaningful. This is documented on BatchingMaxPublishDelayMs, where zero reads back as the 10ms default rather than removing the linger.

batchingConfig requires Apache Pulsar 4.1.0 or later — the field does not exist on ProducerConfig in the 4.0.x line. The field is omitted when nil, so requests to earlier brokers are byte-identical to before this change.

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 batching_config_test.go covering the constructor defaults, JSON serialization across zero-value / disabled / defaults / fully-populated cases, a round trip, and that an absent field decodes as nil while an explicit zero does not.
• Two ProducerConfig tests asserting batchingConfig is absent from the payload when nil — so requests to pre-4.1.0 brokers are unchanged — and present and correctly decoded when set.
• Verified end to end against a real Apache Pulsar 4.1.0 standalone broker: a function created with --producer-config '{"compressionType":"ZSTD","batchingConfig":{"enabled":true,"batchingMaxPublishDelayMs":0,"batchingMaxMessages":100,"batchBuilder":"KEY_BASED"}}' decodes through these types with Enabled=true, BatchBuilder="KEY_BASED", BatchingMaxMessages=100, and BatchingMaxBytes / RoundRobinRouterBatchingPartitionSwitchFrequency as nil rather than 0.
• That same run is where the zero-is-not-off behaviour was confirmed: the broker returned batchingMaxPublishDelayMs: 10 for the zero that was sent.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints

Additive only: a new BatchingConfig type, a new NewBatchingConfig() constructor, a new DefaultBatchingMaxPublishDelayMs constant, and a new nil-by-default field on ProducerConfig. No existing field changes type or serialization.