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

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 

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:

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):

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: