From 5947a5950ddde558df931065026c15d427c4a68e Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 19 Aug 2026 18:05:26 -0700 Subject: [PATCH] [fix][fn] Honour producerSpec batching configuration in the Go 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 Go runtime never reads it. `getProducer` hardcodes `BatchingMaxPublishDelay: time.Millisecond * 10` and never sets `DisableBatching`, so `batchingSpec` arrives in the instance -- the generated bindings in `pb/Function.pb.go` already carry it -- and is silently dropped. Every Go function is pinned to a 10ms publish-latency floor that no configuration can change, and a function configured with `batchingConfig.enabled: false` still batches. `maxPendingMessages` was ignored as well. Fixes #26391 ### Modifications - Add `pf/producerConfig.go` with `producerOptionsFromSpec()`, translating a `ProducerSpec` into `pulsar.ProducerOptions`. - Use it in `getProducer()`, which serves both the sink producer and the producers behind `context.NewOutputMessage()`, so one call site covers both. The translation follows the same rules as the Java runtime: - Unset or non-positive spec fields are left at their zero value so the client default applies. - A nil spec, or a spec with no `BatchingSpec`, keeps batching enabled with a 10ms delay, matching `BatchingUtils.convertFromSpec(nil)`. Existing deployments are unaffected. - `BatchingSpec.BatchBuilder` overrides `ProducerSpec.BatchBuilder`, matching the order in which `ProducerBuilderFactory` applies them. The existing compression and batchBuilder handling moves into the same helper, so `getProducer` no longer mixes configuration translation with producer creation. `RoundRobinRouterBatchingPartitionSwitchFrequency` and `MaxPendingMessagesAcrossPartitions` have no equivalent in the Go client and are ignored. `DisableBlockIfQueueFull` stays false (the producer blocks); 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. No proto regeneration is needed; `pb/Function.pb.go` already contains `BatchingSpec` and `ProducerSpec.GetBatchingSpec()`. ### Verifying this change 14 unit tests added in `pf/producerConfig_test.go`: 10 covering the spec-to-options translation directly, and 4 driving `getProducer` through a fake `pulsar.Client` that captures the options, including a case for a `context.NewOutputMessage()` topic and explicit backwards-compatibility assertions for a function with no producerSpec. 12 of them fail against the unfixed runtime. `go build ./...`, `go test ./pf/...` and `golangci-lint run -c ./golangci.yml ./pf` (v2.12.2, as CI runs it) all pass. --- pulsar-function-go/pf/instance.go | 43 +--- pulsar-function-go/pf/producerConfig.go | 134 ++++++++++ pulsar-function-go/pf/producerConfig_test.go | 253 +++++++++++++++++++ 3 files changed, 397 insertions(+), 33 deletions(-) create mode 100644 pulsar-function-go/pf/producerConfig.go create mode 100644 pulsar-function-go/pf/producerConfig_test.go diff --git a/pulsar-function-go/pf/instance.go b/pulsar-function-go/pf/instance.go index 2cdfc8a6e9497..0f46ea14238f7 100644 --- a/pulsar-function-go/pf/instance.go +++ b/pulsar-function-go/pf/instance.go @@ -261,39 +261,16 @@ func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) { gi.context.instanceConf.funcDetails.Namespace, gi.context.instanceConf.funcDetails.Name), gi.context.instanceConf.instanceID) - batchBuilderType := pulsar.DefaultBatchBuilder - - compressionType := pulsar.LZ4 - if gi.context.instanceConf.funcDetails.Sink.ProducerSpec != nil { - switch gi.context.instanceConf.funcDetails.Sink.ProducerSpec.CompressionType { - case pb.CompressionType_NONE: - compressionType = pulsar.NoCompression - case pb.CompressionType_ZLIB: - compressionType = pulsar.ZLib - case pb.CompressionType_ZSTD: - compressionType = pulsar.ZSTD - default: - compressionType = pulsar.LZ4 // go doesn't support SNAPPY yet - } - - batchBuilder := gi.context.instanceConf.funcDetails.Sink.ProducerSpec.BatchBuilder - if batchBuilder != "" { - if batchBuilder == "KEY_BASED" { - batchBuilderType = pulsar.KeyBasedBatchBuilder - } - } - } - - producer, err := gi.client.CreateProducer(pulsar.ProducerOptions{ - Topic: topicName, - Properties: properties, - CompressionType: compressionType, - BatchingMaxPublishDelay: time.Millisecond * 10, - BatcherBuilderType: batchBuilderType, - SendTimeout: 0, - // Set send timeout to be infinity to prevent potential deadlock with consumer - // that might happen when consumer is blocked due to unacked messages - }) + // Compression and batching come from the function's producerSpec; everything else is fixed by + // the runtime. + options := producerOptionsFromSpec(gi.context.instanceConf.funcDetails.Sink.ProducerSpec) + options.Topic = topicName + options.Properties = properties + // Set send timeout to be infinity to prevent potential deadlock with consumer + // that might happen when consumer is blocked due to unacked messages + options.SendTimeout = 0 + + producer, err := gi.client.CreateProducer(options) if err != nil { gi.stats.incrTotalSysExceptions(err) log.Errorf("create producer error:%s", err.Error()) diff --git a/pulsar-function-go/pf/producerConfig.go b/pulsar-function-go/pf/producerConfig.go new file mode 100644 index 0000000000000..aacc0713ae46a --- /dev/null +++ b/pulsar-function-go/pf/producerConfig.go @@ -0,0 +1,134 @@ +// +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +package pf + +import ( + "time" + + "github.com/apache/pulsar-client-go/pulsar" + + pb "github.com/apache/pulsar/pulsar-function-go/pb" +) + +const ( + // defaultCompressionType and defaultBatchingMaxPublishDelay are the settings a function gets when + // it carries no producer configuration. They mirror the base defaults of the Java runtime's + // ProducerBuilderFactory (enableBatching(true), batchingMaxPublishDelay(10ms)) so that a function + // behaves the same on either runtime when nothing is configured. + defaultCompressionType = pulsar.LZ4 + defaultBatchingMaxPublishDelay = 10 * time.Millisecond + keyBasedBatchBuilder = "KEY_BASED" +) + +// batcherBuilderType translates a batchBuilder name from a function's ProducerSpec into a +// pulsar.BatcherBuilderType. Anything other than "KEY_BASED" maps to the default batcher, matching +// the Java runtime. +func batcherBuilderType(batchBuilder string) pulsar.BatcherBuilderType { + if batchBuilder == keyBasedBatchBuilder { + return pulsar.KeyBasedBatchBuilder + } + return pulsar.DefaultBatchBuilder +} + +// producerOptionsFromSpec builds the producer options a function's sink ProducerSpec configures. +// +// It returns only the settings the spec owns; the caller fills in the rest (topic, properties, send +// timeout, ...). The mapping is: +// +// ProducerSpec field pulsar.ProducerOptions field +// -------------------------------------- ---------------------------------- +// CompressionType CompressionType +// BatchBuilder BatcherBuilderType +// MaxPendingMessages MaxPendingMessages +// BatchingSpec.Enabled DisableBatching (inverted) +// BatchingSpec.BatchingMaxPublishDelayMs BatchingMaxPublishDelay +// BatchingSpec.BatchingMaxMessages BatchingMaxMessages +// BatchingSpec.BatchingMaxBytes BatchingMaxSize +// BatchingSpec.BatchBuilder BatcherBuilderType (wins over BatchBuilder above) +// +// Three rules keep this aligned with the Java runtime (ProducerBuilderFactory and BatchingUtils): +// +// - A field that is unset or non-positive in the spec is left at its zero value, so the client's +// own default applies rather than an explicit zero being pushed onto the producer. +// - A nil spec, or a spec with no BatchingSpec, yields the backwards-compatible defaults: batching +// enabled with a 10ms maximum publish delay. This mirrors BatchingUtils.convertFromSpec(nil) and +// 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 and +// ProducerSpec.MaxPendingMessagesAcrossPartitions have no equivalent in the Go client and are +// ignored. DisableBlockIfQueueFull is left false (the producer blocks), matching the Java runtime, +// which hardcodes blockIfQueueFull(true) and exposes no configuration for it. +func producerOptionsFromSpec(spec *pb.ProducerSpec) pulsar.ProducerOptions { + options := pulsar.ProducerOptions{ + CompressionType: defaultCompressionType, + BatchingMaxPublishDelay: defaultBatchingMaxPublishDelay, + BatcherBuilderType: pulsar.DefaultBatchBuilder, + } + + if spec == nil { + return options + } + + switch spec.CompressionType { + case pb.CompressionType_NONE: + options.CompressionType = pulsar.NoCompression + case pb.CompressionType_ZLIB: + options.CompressionType = pulsar.ZLib + case pb.CompressionType_ZSTD: + options.CompressionType = pulsar.ZSTD + default: + // the Go client does not support SNAPPY yet, so LZ4 covers both LZ4 and SNAPPY + options.CompressionType = pulsar.LZ4 + } + + // 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 spec.BatchBuilder != "" { + options.BatcherBuilderType = batcherBuilderType(spec.BatchBuilder) + } + + if spec.MaxPendingMessages > 0 { + options.MaxPendingMessages = int(spec.MaxPendingMessages) + } + + batchingSpec := spec.GetBatchingSpec() + if batchingSpec == nil { + return options + } + + options.DisableBatching = !batchingSpec.Enabled + if batchingSpec.BatchingMaxPublishDelayMs > 0 { + options.BatchingMaxPublishDelay = time.Duration(batchingSpec.BatchingMaxPublishDelayMs) * time.Millisecond + } + if batchingSpec.BatchingMaxMessages > 0 { + options.BatchingMaxMessages = uint(batchingSpec.BatchingMaxMessages) + } + if batchingSpec.BatchingMaxBytes > 0 { + options.BatchingMaxSize = uint(batchingSpec.BatchingMaxBytes) + } + if batchingSpec.BatchBuilder != "" { + options.BatcherBuilderType = batcherBuilderType(batchingSpec.BatchBuilder) + } + + return options +} diff --git a/pulsar-function-go/pf/producerConfig_test.go b/pulsar-function-go/pf/producerConfig_test.go new file mode 100644 index 0000000000000..ed31fee0a90db --- /dev/null +++ b/pulsar-function-go/pf/producerConfig_test.go @@ -0,0 +1,253 @@ +// +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +package pf + +import ( + "testing" + "time" + + "github.com/apache/pulsar-client-go/pulsar" + "github.com/stretchr/testify/assert" + + pb "github.com/apache/pulsar/pulsar-function-go/pb" +) + +// A function with no producer configuration must keep the behaviour it had before batching became +// configurable: batching on, 10ms maximum publish delay, LZ4, default batcher. +func TestProducerOptionsFromSpec_NilSpecUsesDefaults(t *testing.T) { + options := producerOptionsFromSpec(nil) + + assert.False(t, options.DisableBatching) + assert.Equal(t, 10*time.Millisecond, options.BatchingMaxPublishDelay) + assert.Equal(t, pulsar.LZ4, options.CompressionType) + assert.Equal(t, pulsar.DefaultBatchBuilder, options.BatcherBuilderType) + assert.Zero(t, options.BatchingMaxMessages) + assert.Zero(t, options.BatchingMaxSize) + assert.Zero(t, options.MaxPendingMessages) +} + +func TestProducerOptionsFromSpec_NoBatchingSpecUsesDefaults(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{ + CompressionType: pb.CompressionType_ZSTD, + }) + + assert.False(t, options.DisableBatching) + assert.Equal(t, 10*time.Millisecond, options.BatchingMaxPublishDelay) + assert.Equal(t, pulsar.ZSTD, options.CompressionType) + assert.Zero(t, options.BatchingMaxMessages) +} + +func TestProducerOptionsFromSpec_BatchingCanBeDisabled(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{ + BatchingSpec: &pb.BatchingSpec{Enabled: false}, + }) + + assert.True(t, options.DisableBatching) + // the default delay is still set; it is inert while batching is off + assert.Equal(t, 10*time.Millisecond, options.BatchingMaxPublishDelay) +} + +func TestProducerOptionsFromSpec_FullBatchingSpecIsTranslated(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{ + BatchingSpec: &pb.BatchingSpec{ + Enabled: true, + BatchingMaxPublishDelayMs: 1, + BatchingMaxMessages: 500, + BatchingMaxBytes: 65536, + BatchBuilder: "KEY_BASED", + }, + }) + + assert.False(t, options.DisableBatching) + assert.Equal(t, 1*time.Millisecond, options.BatchingMaxPublishDelay) + assert.Equal(t, uint(500), options.BatchingMaxMessages) + assert.Equal(t, uint(65536), options.BatchingMaxSize) + assert.Equal(t, pulsar.KeyBasedBatchBuilder, options.BatcherBuilderType) +} + +// An explicit zero means "unset" in the protobuf, so the client default must apply rather than a +// literal zero being pushed onto the producer. +func TestProducerOptionsFromSpec_NonPositiveValuesFallBackToClientDefaults(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{ + MaxPendingMessages: 0, + BatchingSpec: &pb.BatchingSpec{ + Enabled: true, + BatchingMaxPublishDelayMs: 0, + BatchingMaxMessages: 0, + BatchingMaxBytes: 0, + }, + }) + + assert.Equal(t, 10*time.Millisecond, options.BatchingMaxPublishDelay) + assert.Zero(t, options.BatchingMaxMessages) + assert.Zero(t, options.BatchingMaxSize) + assert.Zero(t, options.MaxPendingMessages) +} + +func TestProducerOptionsFromSpec_MaxPendingMessagesIsTranslated(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{MaxPendingMessages: 2000}) + + assert.Equal(t, 2000, options.MaxPendingMessages) +} + +func TestProducerOptionsFromSpec_ProducerSpecBatchBuilderIsHonoured(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{BatchBuilder: "KEY_BASED"}) + + assert.Equal(t, pulsar.KeyBasedBatchBuilder, options.BatcherBuilderType) +} + +// The Java runtime applies BatchingSpec.batchBuilder after ProducerSpec.batchBuilder +// (ProducerBuilderFactory), so the nested value must win. +func TestProducerOptionsFromSpec_BatchingSpecBatchBuilderOverridesProducerSpec(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{ + BatchBuilder: "KEY_BASED", + BatchingSpec: &pb.BatchingSpec{Enabled: true, BatchBuilder: "DEFAULT"}, + }) + + assert.Equal(t, pulsar.DefaultBatchBuilder, options.BatcherBuilderType) +} + +func TestProducerOptionsFromSpec_UnknownBatchBuilderFallsBackToDefault(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{BatchBuilder: "SOMETHING_ELSE"}) + + assert.Equal(t, pulsar.DefaultBatchBuilder, options.BatcherBuilderType) +} + +func TestProducerOptionsFromSpec_CompressionTypeIsTranslated(t *testing.T) { + cases := []struct { + name string + spec pb.CompressionType + expected pulsar.CompressionType + }{ + {"LZ4", pb.CompressionType_LZ4, pulsar.LZ4}, + {"NONE", pb.CompressionType_NONE, pulsar.NoCompression}, + {"ZLIB", pb.CompressionType_ZLIB, pulsar.ZLib}, + {"ZSTD", pb.CompressionType_ZSTD, pulsar.ZSTD}, + // the Go client has no SNAPPY support, so it falls back to LZ4 + {"SNAPPY", pb.CompressionType_SNAPPY, pulsar.LZ4}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + options := producerOptionsFromSpec(&pb.ProducerSpec{CompressionType: tc.spec}) + assert.Equal(t, tc.expected, options.CompressionType) + }) + } +} + +// fakePulsarClient records the options getProducer builds so the wiring, and not just the +// translation, is covered. +type fakePulsarClient struct { + pulsar.Client + capturedOptions pulsar.ProducerOptions +} + +func (c *fakePulsarClient) CreateProducer(options pulsar.ProducerOptions) (pulsar.Producer, error) { + c.capturedOptions = options + return nil, nil +} + +func newProducerTestInstance(client pulsar.Client, producerSpec *pb.ProducerSpec) *goInstance { + instance := &goInstance{ + client: client, + context: &FunctionContext{ + instanceConf: &instanceConf{ + instanceID: 0, + funcDetails: pb.FunctionDetails{ + Tenant: "test-tenant", + Namespace: "test-namespace", + Name: "test-function", + Sink: &pb.SinkSpec{ + Topic: "test-sink-topic", + ProducerSpec: producerSpec, + }, + }, + }, + }, + stats: NewStatWithLabelValues("", "", "", "", "", ""), + } + return instance +} + +func TestGetProducer_DefaultsAreUnchangedWithoutAProducerSpec(t *testing.T) { + client := &fakePulsarClient{} + instance := newProducerTestInstance(client, nil) + + _, err := instance.getProducer("test-sink-topic") + + assert.Nil(t, err) + assert.Equal(t, "test-sink-topic", client.capturedOptions.Topic) + assert.False(t, client.capturedOptions.DisableBatching) + assert.Equal(t, 10*time.Millisecond, client.capturedOptions.BatchingMaxPublishDelay) + assert.Equal(t, pulsar.LZ4, client.capturedOptions.CompressionType) + assert.Equal(t, time.Duration(0), client.capturedOptions.SendTimeout) + assert.NotEmpty(t, client.capturedOptions.Properties) +} + +func TestGetProducer_BatchingSpecReachesTheProducer(t *testing.T) { + client := &fakePulsarClient{} + instance := newProducerTestInstance(client, &pb.ProducerSpec{ + MaxPendingMessages: 500, + BatchingSpec: &pb.BatchingSpec{ + Enabled: true, + BatchingMaxPublishDelayMs: 2, + BatchingMaxMessages: 100, + BatchingMaxBytes: 4096, + }, + }) + + _, err := instance.getProducer("test-sink-topic") + + assert.Nil(t, err) + assert.False(t, client.capturedOptions.DisableBatching) + assert.Equal(t, 2*time.Millisecond, client.capturedOptions.BatchingMaxPublishDelay) + assert.Equal(t, uint(100), client.capturedOptions.BatchingMaxMessages) + assert.Equal(t, uint(4096), client.capturedOptions.BatchingMaxSize) + assert.Equal(t, 500, client.capturedOptions.MaxPendingMessages) +} + +func TestGetProducer_BatchingCanBeDisabled(t *testing.T) { + client := &fakePulsarClient{} + instance := newProducerTestInstance(client, &pb.ProducerSpec{ + BatchingSpec: &pb.BatchingSpec{Enabled: false}, + }) + + _, err := instance.getProducer("test-sink-topic") + + assert.Nil(t, err) + assert.True(t, client.capturedOptions.DisableBatching) +} + +// getProducer serves both the sink producer and context.NewOutputMessage(), so a producer for +// another topic must be configured from the same spec. +func TestGetProducer_ContextOutputTopicUsesTheSameSpec(t *testing.T) { + client := &fakePulsarClient{} + instance := newProducerTestInstance(client, &pb.ProducerSpec{ + BatchBuilder: "KEY_BASED", + BatchingSpec: &pb.BatchingSpec{Enabled: true, BatchingMaxPublishDelayMs: 3}, + }) + + _, err := instance.getProducer("another-output-topic") + + assert.Nil(t, err) + assert.Equal(t, "another-output-topic", client.capturedOptions.Topic) + assert.Equal(t, 3*time.Millisecond, client.capturedOptions.BatchingMaxPublishDelay) + assert.Equal(t, pulsar.KeyBasedBatchBuilder, client.capturedOptions.BatcherBuilderType) +}