[fix][fn] Honour producerSpec batching configuration in the Python function runtime

[david-streamlio]

Fixes #26390

Motivation

PIP-401 (#23860) made producer batching configurable for Pulsar Functions. The setting travels as ProducerConfig.batchingConfig → ProducerSpec.batchingSpec in FunctionDetails, and the Java runtime applies it in ProducerBuilderFactory.

The Python runtime never reads it. Both producers it creates hardcode batching_enabled=True and batching_max_publish_delay_ms=10 as literals, so batchingSpec arrives in the instance and is silently dropped. Every Python function is pinned to a 10ms publish-latency floor that no configuration can change.

Below roughly one message per 10ms per instance, that inverts the point of batching: every message opens a batch, waits the full linger for companions that never arrive, and flushes alone. At 27 msg/s per instance, messages arrive every ~37ms, so each one pays a flat 10ms tax while every batch still contains exactly one message.

maxPendingMessages and maxPendingMessagesAcrossPartitions were ignored too, and context.publish() additionally ignored batchBuilder, which the sink producer already honoured.

Modifications

• Add util.producer_config_from_spec() / producer_config_from_function_details(), translating a ProducerSpec into Client.create_producer() keyword arguments.
• Apply it in python_instance.setup_producer() (sink producer) and contextimpl.publish() (context.publish() producers).

The translation follows the same rules as the Java runtime:

• Unset or non-positive spec fields are omitted so the client default applies.
• A sink with no producerSpec, or a producerSpec with no batchingSpec, keeps batching enabled with a 10ms delay, matching BatchingUtils.convertFromSpec(null).
• batchingSpec.batchBuilder overrides ProducerSpec.batchBuilder, matching the order ProducerBuilderFactory applies them in.

Omitting batching_type when no batchBuilder is configured is behaviour preserving: the Python client already defaults it to BatchingType.Default, which is what the previous unconditional argument passed.

Out of scope, deliberately:

• block_if_queue_full stays fixed at True. The Java runtime hardcodes blockIfQueueFull(true) too and exposes no configuration for it, so making it configurable means adding a new ProducerSpec field — a config-surface change that belongs in its own PR.
• roundRobinRouterBatchingPartitionSwitchFrequency has no equivalent in the Python client and is ignored.
• The Go runtime has the identical gap, tracked separately in [Functions] Go function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec #26391.

Documentation

This PR updates pulsar-functions/instance/src/main/python/README.md with a ProducerSpec → create_producer() mapping table and the parity rules above, but that is contributor-facing documentation only.

The user-facing docs still need a change in apache/pulsar-site, and I have not made it. Concretely:

• The Pulsar Functions producer/batching configuration reference documents batchingConfig without noting which runtimes honour it. Before this PR it was Java-only in practice; after it, Java and Python. Readers had no way to know their Python function was ignoring the setting, which is what made [Functions] Python function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec #26390 hard to spot in the first place.
• Whatever wording lands should stay accurate for Go, which still ignores batchingSpec ([Functions] Go function runtime hardcodes producer batching config, ignoring producerSpec.batchingSpec #26391). Either scope the note to the runtimes that honour it today, or hold the site change until the Go fix lands and document both at once.

Marking this doc-required rather than doc-complete for that reason. Happy to open the pulsar-site PR as a follow-up — flagging it here so it does not get lost, and so a reviewer can tell me whether they would rather wait for #26391 and do one combined docs update.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• 19 unit tests added to test_python_instance.py, covering the spec-to-kwargs translation directly and asserting the resulting create_producer() call for both producer paths.
• Explicit backwards-compatibility cases assert that a function with no producerSpec still gets batching_enabled=True, batching_max_publish_delay_ms=10, block_if_queue_full=True.
• Reverting the three production files and re-running the suite fails 16 of the 19, confirming the tests pin the new behaviour rather than passing vacuously.
• Full suite: 23/23 pass (4 pre-existing + 19 new), via pulsar-functions/instance/src/scripts/run_python_instance_tests.sh.
• ./gradlew quickCheck passes.

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

If the box was checked, please highlight the changes

• 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
• The admin CLI options
• The metrics
• Anything that affects deployment

No box checked: no configuration surface is added or changed, and defaults are unchanged. This makes the Python runtime honour configuration that already exists and is already accepted from users today.

[david-streamlio]

The doc-required label on this PR is stale — the documentation has already merged.

apache/pulsar-site#1215 ("Document producerConfig.batchingConfig for Pulsar Functions") merged on 25 Aug and covers exactly the surface this PR implements in the Python runtime. Nothing further is owed on the docs side, so I have relabelled to doc-complete; happy to revert if a reviewer sees it differently.

Flagging it because doc-required reads as "the author still owes work", and this PR has had no review activity since it was opened on 20 Aug despite being green.

For context on the impact: at low throughput (~27 msg/s per instance) a function instance never fills a batch, so every message waits the full hardcoded batching delay and then flushes alone — a flat per-message latency tax that the producerSpec settings are meant to let operators tune away. Today those settings are silently ignored by the runtime.

[freeznet]

Verified the ProducerSpec/BatchingSpec mapping against FunctionConfigUtils, BatchingUtils, ProducerBuilderFactory, and pulsar-client 3.13.0. Both producer paths preserve the existing defaults and apply the configured limits and batcher precedence. The Python runtime tests pass locally.