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

pulsaradmin/pkg/utils/batching_config.go

@@ -0,0 +1,77 @@
+// 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 utils
+
+// DefaultBatchingMaxPublishDelayMs is the maximum publish delay the broker applies when a producer
+// has no batching configuration. It mirrors the default in the Java runtime's
+// BatchingUtils.convertFromSpec(null).
+const DefaultBatchingMaxPublishDelayMs = 10
+
+// BatchingConfig is the producer batching configuration for a Pulsar Function, Source or Sink,
+// introduced by PIP-401.
+//
+// It requires a broker running Apache Pulsar 4.1.0 or later. Earlier releases have no batchingConfig
+// field on producerConfig and ignore this value, leaving the producer on its built-in defaults.
+//
+// The pointer fields distinguish "not configured" from an explicit value, matching the boxed
+// Integer fields in the Java model. Note that the broker applies each of them only when it is
+// present and greater than zero, so a nil field and a zero field both defer to the default; zero is
+// not a way to switch a limit off. Use Enabled to turn batching off entirely.
+type BatchingConfig struct {
+	// Enabled reports whether batching is on. It is always serialized, with no omitempty, because
+	// the broker reads it as a primitive boolean with no fallback: BatchingUtils.convert() calls
+	// setEnabled(config.isEnabled()) unconditionally, so a payload that omits the field can leave
+	// batching disabled rather than defaulting to on. Use NewBatchingConfig to start from the same
+	// defaults the broker applies when no configuration is present.
+	Enabled bool `json:"enabled" yaml:"enabled"`
+
+	// BatchingMaxPublishDelayMs is the batching linger in milliseconds.
+	//
+	// Zero does not disable the linger. BatchingUtils.convert() applies the value only when it is
+	// greater than zero, so the broker treats zero as unconfigured and falls back to
+	// DefaultBatchingMaxPublishDelayMs. Verified against Pulsar 4.1.0: a request carrying zero reads
+	// back as 10. Set Enabled to false to turn batching off.
+	BatchingMaxPublishDelayMs *int `json:"batchingMaxPublishDelayMs,omitempty" yaml:"batchingMaxPublishDelayMs"`
+
+	//nolint:lll
+	RoundRobinRouterBatchingPartitionSwitchFrequency *int `json:"roundRobinRouterBatchingPartitionSwitchFrequency,omitempty" yaml:"roundRobinRouterBatchingPartitionSwitchFrequency"`
+
+	BatchingMaxMessages *int `json:"batchingMaxMessages,omitempty" yaml:"batchingMaxMessages"`
+
+	BatchingMaxBytes *int `json:"batchingMaxBytes,omitempty" yaml:"batchingMaxBytes"`
+
+	// BatchBuilder selects the batch construction method, either DEFAULT or KEY_BASED. When set, it
+	// takes precedence over ProducerConfig.BatchBuilder, matching the order the Java runtime's
+	// ProducerBuilderFactory applies them in.
+	BatchBuilder string `json:"batchBuilder,omitempty" yaml:"batchBuilder"`
+}
+
+// NewBatchingConfig returns a BatchingConfig holding the defaults the broker applies to a producer
+// with no batching configuration: batching enabled with a 10ms maximum publish delay.
+//
+// Prefer it over a bare BatchingConfig literal when tuning a single field. Go's zero value for
+// Enabled is false, so a literal that sets only, say, BatchingMaxMessages would disable batching
+// rather than cap the batch size.
+func NewBatchingConfig() *BatchingConfig {
+	maxPublishDelayMs := DefaultBatchingMaxPublishDelayMs
+
+	return &BatchingConfig{
+		Enabled:                   true,
+		BatchingMaxPublishDelayMs: &maxPublishDelayMs,
+	}
+}

pulsaradmin/pkg/utils/batching_config_test.go

@@ -0,0 +1,155 @@
+// 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 utils
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func intPtr(value int) *int {
+	return &value
+}
+
+func TestNewBatchingConfig(t *testing.T) {
+	config := NewBatchingConfig()
+
+	require.NotNil(t, config)
+	assert.IsType(t, &BatchingConfig{}, config)
+
+	// The broker applies these when a producer carries no batching configuration
+	// (BatchingUtils.convertFromSpec(null)), so the constructor is a no-op change in behaviour.
+	assert.True(t, config.Enabled)
+	require.NotNil(t, config.BatchingMaxPublishDelayMs)
+	assert.Equal(t, DefaultBatchingMaxPublishDelayMs, *config.BatchingMaxPublishDelayMs)
+
+	assert.Nil(t, config.RoundRobinRouterBatchingPartitionSwitchFrequency)
+	assert.Nil(t, config.BatchingMaxMessages)
+	assert.Nil(t, config.BatchingMaxBytes)
+	assert.Empty(t, config.BatchBuilder)
+}
+
+func TestBatchingConfigJSONSerialization(t *testing.T) {
+	tests := []struct {
+		name     string
+		config   BatchingConfig
+		expected string
+	}{
+		{
+			// enabled carries no omitempty: BatchingUtils.convert() reads it as a primitive with no
+			// fallback, so a payload omitting it can leave batching off.
+			name:     "zero value still emits enabled",
+			config:   BatchingConfig{},
+			expected: `{"enabled":false}`,
+		},
+		{
+			name:     "explicitly disabled",
+			config:   BatchingConfig{Enabled: false},
+			expected: `{"enabled":false}`,
+		},
+		{
+			name:     "constructor defaults",
+			config:   *NewBatchingConfig(),
+			expected: `{"enabled":true,"batchingMaxPublishDelayMs":10}`,
+		},
+		{
+			// Serialized faithfully, but the broker ignores a non-positive delay and falls back to
+			// its 10ms default; Enabled is what turns batching off. Asserted so the wire format
+			// stays honest about what the caller asked for.
+			name: "explicit zero max publish delay is serialized",
+			config: BatchingConfig{
+				Enabled:                   true,
+				BatchingMaxPublishDelayMs: intPtr(0),
+			},
+			expected: `{"enabled":true,"batchingMaxPublishDelayMs":0}`,
+		},
+		{
+			name: "all fields",
+			config: BatchingConfig{
+				Enabled:                   true,
+				BatchingMaxPublishDelayMs: intPtr(5),
+				RoundRobinRouterBatchingPartitionSwitchFrequency: intPtr(20),
+				BatchingMaxMessages: intPtr(100),
+				BatchingMaxBytes:    intPtr(131072),
+				BatchBuilder:        "KEY_BASED",
+			},
+			//nolint:lll
+			expected: `{"enabled":true,"batchingMaxPublishDelayMs":5,"roundRobinRouterBatchingPartitionSwitchFrequency":20,"batchingMaxMessages":100,"batchingMaxBytes":131072,"batchBuilder":"KEY_BASED"}`,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			data, err := json.Marshal(test.config)
+			require.NoError(t, err)
+			assert.JSONEq(t, test.expected, string(data))
+		})
+	}
+}
+
+func TestBatchingConfigRoundTrip(t *testing.T) {
+	original := BatchingConfig{
+		Enabled:                   true,
+		BatchingMaxPublishDelayMs: intPtr(0),
+		BatchingMaxMessages:       intPtr(100),
+		BatchBuilder:              "KEY_BASED",
+	}
+
+	data, err := json.Marshal(original)
+	require.NoError(t, err)
+
+	var decoded BatchingConfig
+	require.NoError(t, json.Unmarshal(data, &decoded))
+	assert.Equal(t, original, decoded)
+}
+
+func TestBatchingConfigUnmarshalDistinguishesUnsetFromZero(t *testing.T) {
+	var absent BatchingConfig
+	require.NoError(t, json.Unmarshal([]byte(`{"enabled":true}`), &absent))
+	assert.Nil(t, absent.BatchingMaxMessages, "an absent field must stay nil")
+
+	var zero BatchingConfig
+	require.NoError(t, json.Unmarshal([]byte(`{"enabled":true,"batchingMaxMessages":0}`), &zero))
+	require.NotNil(t, zero.BatchingMaxMessages, "an explicit zero must not read as unset")
+	assert.Equal(t, 0, *zero.BatchingMaxMessages)
+}
+
+func TestProducerConfigOmitsBatchingConfigWhenUnset(t *testing.T) {
+	// Requests to brokers older than 4.1.0 must be byte-identical to before this field existed.
+	data, err := json.Marshal(ProducerConfig{})
+	require.NoError(t, err)
+	assert.NotContains(t, string(data), "batchingConfig")
+}
+
+func TestProducerConfigIncludesBatchingConfigWhenSet(t *testing.T) {
+	config := ProducerConfig{BatchingConfig: NewBatchingConfig()}
+
+	data, err := json.Marshal(config)
+	require.NoError(t, err)
+
+	var decoded ProducerConfig
+	require.NoError(t, json.Unmarshal(data, &decoded))
+
+	require.NotNil(t, decoded.BatchingConfig)
+	assert.True(t, decoded.BatchingConfig.Enabled)
+	require.NotNil(t, decoded.BatchingConfig.BatchingMaxPublishDelayMs)
+	assert.Equal(t, DefaultBatchingMaxPublishDelayMs, *decoded.BatchingConfig.BatchingMaxPublishDelayMs)
+}

pulsaradmin/pkg/utils/producer_config.go

@@ -26,4 +26,9 @@ type ProducerConfig struct {
 	CryptoConfig            *CryptoConfig `json:"cryptoConfig" yaml:"cryptoConfig"`
 	BatchBuilder            string        `json:"batchBuilder" yaml:"batchBuilder"`
 	CompressionType         string        `json:"compressionType" yaml:"compressionType"`
+
+	// BatchingConfig requires Apache Pulsar 4.1.0 or later, which added batchingConfig to
+	// producerConfig via PIP-401. It is omitted when nil so that requests to earlier brokers are
+	// unchanged.
+	BatchingConfig *BatchingConfig `json:"batchingConfig,omitempty" yaml:"batchingConfig"`
 }