From 3bb4426eb4df2c68b33f2b426b003f38e0ef609d Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 19 Aug 2026 17:53:22 -0700 Subject: [PATCH] [fix][fn] Honour producerSpec batching configuration in the Python function runtime ### 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, and a function whose per-instance rate is below one message per 10ms pays that delay on every message 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. Fixes #26390 ### 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 in `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)`. Existing deployments are unaffected. - `batchingSpec.batchBuilder` overrides `ProducerSpec.batchBuilder`, matching the order in which `ProducerBuilderFactory` applies them. 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. `roundRobinRouterBatchingPartitionSwitchFrequency` has no equivalent in the Python client and is ignored. `block_if_queue_full` stays fixed at `True`; the Java runtime hardcodes `blockIfQueueFull(true)` as well and exposes no configuration for it, so making it configurable would need a new proto field and belongs in a separate change. ### Verifying this change 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, including the unchanged defaults. 16 of them fail against the unfixed runtime. --- .../instance/src/main/python/README.md | 38 ++++ .../instance/src/main/python/contextimpl.py | 8 +- .../src/main/python/python_instance.py | 14 +- .../instance/src/main/python/util.py | 72 ++++++ .../src/test/python/test_python_instance.py | 211 ++++++++++++++++++ 5 files changed, 330 insertions(+), 13 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/README.md b/pulsar-functions/instance/src/main/python/README.md index bbbe149afec15..6465d88513115 100644 --- a/pulsar-functions/instance/src/main/python/README.md +++ b/pulsar-functions/instance/src/main/python/README.md @@ -1,5 +1,43 @@ # Pulsar Functions Python Runtime +### Producer configuration + +Both producers the runtime creates — the sink (output topic) producer in `python_instance.py` and the +producers behind `context.publish()` in `contextimpl.py` — are configured from the `producerSpec` of +the function's sink, which reaches the instance inside the `FunctionDetails` protobuf. The +translation lives in `util.producer_config_from_function_details()`. + +| `ProducerSpec` field | `Client.create_producer()` keyword | +|---|---| +| `maxPendingMessages` | `max_pending_messages` | +| `maxPendingMessagesAcrossPartitions` | `max_pending_messages_across_partitions` | +| `batchBuilder` | `batching_type` | +| `compressionType` | `compression_type` (sink producer only; `context.publish()` takes a per-call value) | +| `cryptoSpec` | `crypto_key_reader`, `encryption_key` (sink producer only) | +| `batchingSpec.enabled` | `batching_enabled` | +| `batchingSpec.batchingMaxPublishDelayMs` | `batching_max_publish_delay_ms` | +| `batchingSpec.batchingMaxMessages` | `batching_max_messages` | +| `batchingSpec.batchingMaxBytes` | `batching_max_allowed_size_in_bytes` | +| `batchingSpec.batchBuilder` | `batching_type` (takes precedence over `ProducerSpec.batchBuilder`) | + +These are the same settings a user configures through `producerConfig` on the function config, which +PIP-401 extended with `batchingConfig`. + +A few rules keep the behaviour aligned with the Java runtime +(`ProducerBuilderFactory` and `BatchingUtils`): + +- **A field that is unset or non-positive in the spec is left out**, so the Python client's own + default applies rather than an explicit zero. +- **A sink with no `producerSpec`, or a `producerSpec` with no `batchingSpec`, gets batching enabled + with a 10ms maximum publish delay.** This is the long-standing default and must not change: it is + what functions written before batching became configurable already run with. +- **`batchingSpec.batchBuilder` wins over `ProducerSpec.batchBuilder`**, because the Java runtime + applies them in that order. + +`batchingSpec.roundRobinRouterBatchingPartitionSwitchFrequency` has no equivalent in the Python +client and is ignored. `block_if_queue_full` is fixed at `True`, matching the Java runtime, which +also hardcodes `blockIfQueueFull(true)` and exposes no configuration for it. + ### Updating Protobuf and gRPC generated stubs When using generated Protobuf and gRPC stubs (`*_pb2.py`, `*_pb2_gprc.py`), the generated code should be diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py index 826ad65b2e024..f885f4fcb49b6 100755 --- a/pulsar-functions/instance/src/main/python/contextimpl.py +++ b/pulsar-functions/instance/src/main/python/contextimpl.py @@ -172,17 +172,19 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", p if compression_type is not None: pulsar_compression_type = compression_type if topic_name not in self.publish_producers: + # Honour the batching / pending-queue settings configured on the function's producerSpec, the + # same ones the sink producer uses, so that context.publish() is not pinned to the defaults. + producer_config = util.producer_config_from_function_details(self.instance_config.function_details) self.publish_producers[topic_name] = self.pulsar_client.create_producer( topic_name, block_if_queue_full=True, - batching_enabled=True, - batching_max_publish_delay_ms=10, compression_type=pulsar_compression_type, properties=util.get_properties(util.getFullyQualifiedFunctionName( self.instance_config.function_details.tenant, self.instance_config.function_details.namespace, self.instance_config.function_details.name), - self.instance_config.instance_id) + self.instance_config.instance_id), + **producer_config ) if serde_class_name not in self.publish_serializers: diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 5c57dfef79008..61322049f2128 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -372,12 +372,7 @@ def setup_producer(self, producer_name=None): len(self.instance_config.function_details.sink.topic) > 0: Log.debug("Setting up producer for topic %s" % self.instance_config.function_details.sink.topic) - batch_type = pulsar.BatchingType.Default - if self.instance_config.function_details.sink.producerSpec.batchBuilder != None and \ - len(self.instance_config.function_details.sink.producerSpec.batchBuilder) > 0: - batch_builder = self.instance_config.function_details.sink.producerSpec.batchBuilder - if batch_builder == "KEY_BASED": - batch_type = pulsar.BatchingType.KeyBased + producer_config = util.producer_config_from_function_details(self.instance_config.function_details) self.output_schema = self.get_schema(self.instance_config.function_details.sink.schemaType, self.instance_config.function_details.sink.typeClassName, @@ -403,9 +398,6 @@ def setup_producer(self, producer_name=None): schema=self.output_schema, producer_name=producer_name, block_if_queue_full=True, - batching_enabled=True, - batching_type=batch_type, - batching_max_publish_delay_ms=10, compression_type=compression_type, # set send timeout to be infinity to prevent potential deadlock with consumer # that might happen when consumer is blocked due to unacked messages @@ -417,7 +409,9 @@ def setup_producer(self, producer_name=None): self.instance_config.function_details.tenant, self.instance_config.function_details.namespace, self.instance_config.function_details.name), - self.instance_config.instance_id) + self.instance_config.instance_id), + # batching / pending-queue settings configured on the function's producerSpec + **producer_config ) def setup_state(self): diff --git a/pulsar-functions/instance/src/main/python/util.py b/pulsar-functions/instance/src/main/python/util.py index 7b2a2d7b4d172..a113618f7f9be 100755 --- a/pulsar-functions/instance/src/main/python/util.py +++ b/pulsar-functions/instance/src/main/python/util.py @@ -28,6 +28,7 @@ import configparser from threading import Timer +import pulsar from pulsar.functions import serde import log @@ -82,6 +83,77 @@ def getFullyQualifiedInstanceId(tenant, namespace, name, instance_id): def get_properties(fullyQualifiedName, instanceId): return {"application": "pulsar-function", "id": str(fullyQualifiedName), "instance_id": str(instanceId)} +# Defaults applied when a function carries no producer configuration. These deliberately mirror the +# base defaults in the Java runtime's ProducerBuilderFactory (blockIfQueueFull(true), +# enableBatching(true), batchingMaxPublishDelay(10ms)) so that a function behaves the same on either +# runtime when nothing is configured. +DEFAULT_BATCHING_ENABLED = True +DEFAULT_BATCHING_MAX_PUBLISH_DELAY_MS = 10 + +def batching_type_from_batch_builder(batch_builder): + """Translate a batchBuilder name from the function's ProducerSpec into a pulsar.BatchingType. + + Anything other than "KEY_BASED" maps to the default batcher, matching the Java runtime. + """ + if batch_builder == "KEY_BASED": + return pulsar.BatchingType.KeyBased + return pulsar.BatchingType.Default + +def producer_config_from_spec(producer_spec): + """Translate a ProducerSpec protobuf into keyword arguments for Client.create_producer(). + + Returns the batching and pending-queue settings only; the caller owns everything else (topic, + schema, compression, crypto, properties, ...). Fields that are unset or non-positive in the spec + are left out of the result so that the client's own defaults apply, which is the same rule the + Java runtime follows in ProducerBuilderFactory. + + Passing None (no producerSpec on the sink) yields the backwards-compatible defaults: batching + enabled with a 10ms maximum publish delay. This mirrors BatchingUtils.convertFromSpec(null). + """ + config = { + "batching_enabled": DEFAULT_BATCHING_ENABLED, + "batching_max_publish_delay_ms": DEFAULT_BATCHING_MAX_PUBLISH_DELAY_MS, + } + + if producer_spec is None: + return config + + # batchBuilder lives on the ProducerSpec itself and, since PIP-401, also on the nested + # BatchingSpec. The Java runtime applies the ProducerSpec one first and lets the BatchingSpec one + # override it, so do the same here. + if producer_spec.batchBuilder: + config["batching_type"] = batching_type_from_batch_builder(producer_spec.batchBuilder) + + if producer_spec.maxPendingMessages > 0: + config["max_pending_messages"] = producer_spec.maxPendingMessages + if producer_spec.maxPendingMessagesAcrossPartitions > 0: + config["max_pending_messages_across_partitions"] = producer_spec.maxPendingMessagesAcrossPartitions + + if not producer_spec.HasField("batchingSpec"): + return config + + batching_spec = producer_spec.batchingSpec + config["batching_enabled"] = batching_spec.enabled + if batching_spec.batchingMaxPublishDelayMs > 0: + config["batching_max_publish_delay_ms"] = batching_spec.batchingMaxPublishDelayMs + if batching_spec.batchingMaxMessages > 0: + config["batching_max_messages"] = batching_spec.batchingMaxMessages + if batching_spec.batchingMaxBytes > 0: + config["batching_max_allowed_size_in_bytes"] = batching_spec.batchingMaxBytes + if batching_spec.batchBuilder: + config["batching_type"] = batching_type_from_batch_builder(batching_spec.batchBuilder) + + return config + +def producer_config_from_function_details(function_details): + """Return the producer keyword arguments configured on a function's sink. + + Safe to call for any function: sinks without a producerSpec fall back to the defaults. + """ + if function_details is None or not function_details.sink.HasField("producerSpec"): + return producer_config_from_spec(None) + return producer_config_from_spec(function_details.sink.producerSpec) + def read_config(config_file): """ The content of the configuration file is styled as follows: diff --git a/pulsar-functions/instance/src/test/python/test_python_instance.py b/pulsar-functions/instance/src/test/python/test_python_instance.py index 1e72db8545816..3b20ac4b54563 100644 --- a/pulsar-functions/instance/src/test/python/test_python_instance.py +++ b/pulsar-functions/instance/src/test/python/test_python_instance.py @@ -37,7 +37,9 @@ import Function_pb2 import log import os +import pulsar import unittest +import util class TestContextImpl(unittest.TestCase): @@ -149,3 +151,212 @@ def test_do_not_forward_properties(self): self.assertNotIn("custom-key", kwargs['properties']) self.assertIn("__pfn_input_topic__", kwargs['properties']) + +class TestProducerConfigFromSpec(unittest.TestCase): + """Unit tests for the ProducerSpec -> create_producer() keyword translation.""" + + def test_defaults_when_no_producer_spec(self): + function_details = Function_pb2.FunctionDetails() + config = util.producer_config_from_function_details(function_details) + self.assertEqual(config, { + "batching_enabled": True, + "batching_max_publish_delay_ms": 10, + }) + + def test_defaults_when_producer_spec_has_no_batching_spec(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.compressionType = Function_pb2.CompressionType.Value("ZSTD") + config = util.producer_config_from_function_details(function_details) + self.assertTrue(config["batching_enabled"]) + self.assertEqual(config["batching_max_publish_delay_ms"], 10) + self.assertNotIn("batching_max_messages", config) + self.assertNotIn("batching_type", config) + + def test_batching_can_be_disabled(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.batchingSpec.enabled = False + config = util.producer_config_from_function_details(function_details) + self.assertFalse(config["batching_enabled"]) + # the default delay is still reported; it is inert while batching is off + self.assertEqual(config["batching_max_publish_delay_ms"], 10) + + def test_full_batching_spec_is_translated(self): + function_details = Function_pb2.FunctionDetails() + batching_spec = function_details.sink.producerSpec.batchingSpec + batching_spec.enabled = True + batching_spec.batchingMaxPublishDelayMs = 1 + batching_spec.batchingMaxMessages = 500 + batching_spec.batchingMaxBytes = 65536 + batching_spec.batchBuilder = "KEY_BASED" + config = util.producer_config_from_function_details(function_details) + self.assertEqual(config, { + "batching_enabled": True, + "batching_max_publish_delay_ms": 1, + "batching_max_messages": 500, + "batching_max_allowed_size_in_bytes": 65536, + "batching_type": pulsar.BatchingType.KeyBased, + }) + + def test_non_positive_values_fall_back_to_client_defaults(self): + function_details = Function_pb2.FunctionDetails() + batching_spec = function_details.sink.producerSpec.batchingSpec + batching_spec.enabled = True + batching_spec.batchingMaxPublishDelayMs = 0 + batching_spec.batchingMaxMessages = 0 + batching_spec.batchingMaxBytes = 0 + config = util.producer_config_from_function_details(function_details) + # an explicit zero means "unset" in the protobuf, so the runtime default applies + self.assertEqual(config["batching_max_publish_delay_ms"], 10) + self.assertNotIn("batching_max_messages", config) + self.assertNotIn("batching_max_allowed_size_in_bytes", config) + + def test_pending_message_limits_are_translated(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.maxPendingMessages = 2000 + function_details.sink.producerSpec.maxPendingMessagesAcrossPartitions = 8000 + config = util.producer_config_from_function_details(function_details) + self.assertEqual(config["max_pending_messages"], 2000) + self.assertEqual(config["max_pending_messages_across_partitions"], 8000) + + def test_pending_message_limits_omitted_when_unset(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.batchingSpec.enabled = True + config = util.producer_config_from_function_details(function_details) + self.assertNotIn("max_pending_messages", config) + self.assertNotIn("max_pending_messages_across_partitions", config) + + def test_producer_spec_batch_builder_is_honoured(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.batchBuilder = "KEY_BASED" + config = util.producer_config_from_function_details(function_details) + self.assertEqual(config["batching_type"], pulsar.BatchingType.KeyBased) + + def test_batching_spec_batch_builder_overrides_producer_spec(self): + # the Java runtime applies BatchingSpec.batchBuilder after ProducerSpec.batchBuilder + # (ProducerBuilderFactory), so the nested value must win + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.batchBuilder = "KEY_BASED" + function_details.sink.producerSpec.batchingSpec.enabled = True + function_details.sink.producerSpec.batchingSpec.batchBuilder = "DEFAULT" + config = util.producer_config_from_function_details(function_details) + self.assertEqual(config["batching_type"], pulsar.BatchingType.Default) + + def test_unknown_batch_builder_falls_back_to_default(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.batchBuilder = "SOMETHING_ELSE" + config = util.producer_config_from_function_details(function_details) + self.assertEqual(config["batching_type"], pulsar.BatchingType.Default) + + def test_none_function_details_yields_defaults(self): + config = util.producer_config_from_function_details(None) + self.assertEqual(config, { + "batching_enabled": True, + "batching_max_publish_delay_ms": 10, + }) + + +class TestSinkProducerBatchingConfig(unittest.TestCase): + """The sink (output topic) producer must be built from the function's producerSpec.""" + + def _create_producer_kwargs(self, function_details): + mock_pulsar_client = Mock() + mock_pulsar_client.create_producer.return_value = Mock() + instance = PythonInstance('test_instance', 'test_func', '1.0', function_details, 100, 30, + 'user_code', mock_pulsar_client, Mock(), 'test_cluster', 'test_url', None) + instance.get_schema = Mock(return_value="DEFAULT_SCHEMA") + instance.get_crypto_reader = Mock(return_value=None) + instance.setup_producer() + _, kwargs = mock_pulsar_client.create_producer.call_args + return kwargs + + def test_defaults_are_unchanged_without_a_producer_spec(self): + # backwards compatibility: a function with no producer configuration must keep batching on + # with a 10ms maximum publish delay, exactly as before this was configurable + function_details = Function_pb2.FunctionDetails() + function_details.sink.topic = "test_sink_topic" + kwargs = self._create_producer_kwargs(function_details) + self.assertTrue(kwargs["batching_enabled"]) + self.assertEqual(kwargs["batching_max_publish_delay_ms"], 10) + self.assertTrue(kwargs["block_if_queue_full"]) + + def test_batching_disabled_reaches_the_producer(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.topic = "test_sink_topic" + function_details.sink.producerSpec.batchingSpec.enabled = False + kwargs = self._create_producer_kwargs(function_details) + self.assertFalse(kwargs["batching_enabled"]) + + def test_batching_settings_reach_the_producer(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.topic = "test_sink_topic" + batching_spec = function_details.sink.producerSpec.batchingSpec + batching_spec.enabled = True + batching_spec.batchingMaxPublishDelayMs = 2 + batching_spec.batchingMaxMessages = 100 + batching_spec.batchingMaxBytes = 4096 + function_details.sink.producerSpec.maxPendingMessages = 500 + kwargs = self._create_producer_kwargs(function_details) + self.assertTrue(kwargs["batching_enabled"]) + self.assertEqual(kwargs["batching_max_publish_delay_ms"], 2) + self.assertEqual(kwargs["batching_max_messages"], 100) + self.assertEqual(kwargs["batching_max_allowed_size_in_bytes"], 4096) + self.assertEqual(kwargs["max_pending_messages"], 500) + + def test_key_based_batch_builder_still_reaches_the_producer(self): + # this was the one producerSpec field the sink producer already honoured; keep it working + function_details = Function_pb2.FunctionDetails() + function_details.sink.topic = "test_sink_topic" + function_details.sink.producerSpec.batchBuilder = "KEY_BASED" + kwargs = self._create_producer_kwargs(function_details) + self.assertEqual(kwargs["batching_type"], pulsar.BatchingType.KeyBased) + + +class TestContextPublishBatchingConfig(unittest.TestCase): + """context.publish() producers must be built from the same producerSpec as the sink producer.""" + + def _create_producer_kwargs(self, function_details): + instance_config = InstanceConfig('test_instance_id', 'test_function_id', 'test_function_version', + function_details, 100) + pulsar_client = Mock() + producer = Mock() + producer.send_async = Mock(return_value=None) + pulsar_client.create_producer = Mock(return_value=producer) + context_impl = ContextImpl(instance_config, log.Log, pulsar_client, __file__, None, None, None, None, None) + + msg = Message() + msg.message_id = Mock(return_value="test_message_id") + msg.partition_key = Mock(return_value="test_key") + context_impl.set_current_message_context(msg, "test_topic_name") + context_impl.publish("test_topic_name", "test_message") + + _, kwargs = pulsar_client.create_producer.call_args + return kwargs + + def test_defaults_are_unchanged_without_a_producer_spec(self): + kwargs = self._create_producer_kwargs(Function_pb2.FunctionDetails()) + self.assertTrue(kwargs["batching_enabled"]) + self.assertEqual(kwargs["batching_max_publish_delay_ms"], 10) + self.assertTrue(kwargs["block_if_queue_full"]) + + def test_batching_disabled_reaches_the_producer(self): + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.batchingSpec.enabled = False + kwargs = self._create_producer_kwargs(function_details) + self.assertFalse(kwargs["batching_enabled"]) + + def test_batching_settings_reach_the_producer(self): + function_details = Function_pb2.FunctionDetails() + batching_spec = function_details.sink.producerSpec.batchingSpec + batching_spec.enabled = True + batching_spec.batchingMaxPublishDelayMs = 5 + batching_spec.batchingMaxMessages = 250 + kwargs = self._create_producer_kwargs(function_details) + self.assertEqual(kwargs["batching_max_publish_delay_ms"], 5) + self.assertEqual(kwargs["batching_max_messages"], 250) + + def test_batch_builder_reaches_the_producer(self): + # context.publish() previously ignored batchBuilder entirely + function_details = Function_pb2.FunctionDetails() + function_details.sink.producerSpec.batchBuilder = "KEY_BASED" + kwargs = self._create_producer_kwargs(function_details) + self.assertEqual(kwargs["batching_type"], pulsar.BatchingType.KeyBased)