diff --git a/pulsaradmin/pkg/utils/batching_config.go b/pulsaradmin/pkg/utils/batching_config.go new file mode 100644 index 0000000000..db92e835ec --- /dev/null +++ b/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, + } +} diff --git a/pulsaradmin/pkg/utils/batching_config_test.go b/pulsaradmin/pkg/utils/batching_config_test.go new file mode 100644 index 0000000000..286eb2ffcf --- /dev/null +++ b/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) +} diff --git a/pulsaradmin/pkg/utils/producer_config.go b/pulsaradmin/pkg/utils/producer_config.go index d4c1c3eb6d..cbaf4ac7ed 100644 --- a/pulsaradmin/pkg/utils/producer_config.go +++ b/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"` }