feat(function): expose producer configuration on pulsar_function

docs/resources/function.md

@@ -22,8 +22,10 @@ Manages Pulsar Functions through the Functions Worker API.
 ### Optional
 
 - `auto_ack` (Boolean) Whether to automatically acknowledge messages processed by the function.
+- `batch_builder` (String) BatchBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED.
 - `classname` (String) The class name of the function.
 - `cleanup_subscription` (Boolean) Whether to clean up subscription when the function is deleted.
+- `compression_type` (String) Set the compression type for the producer. Pulsar Functions default to LZ4. Supported compression types are: LZ4, ZLIB, ZSTD, SNAPPY and NONE
 - `cpu` (Number) The CPU that needs to be allocated per function instance
 - `custom_runtime_options` (String) The custom runtime options of the function.
 - `custom_schema_inputs` (Map of String) The custom schema inputs of the function.
@@ -39,6 +41,8 @@ Manages Pulsar Functions through the Functions Worker API.
 - `jar` (String) The path to the jar file.
 - `log_topic` (String) The log topic of the function.
 - `max_message_retries` (Number) The maximum number of times that a message will be retried when the function is configured with `EFFECTIVELY_ONCE` processing guarantees.
+- `max_pending_messages` (Number) The maximum size of a queue holding pending messages
+- `max_pending_messages_across_partitions` (Number) The maximum number of pending messages across partitions
 - `output` (String) The output topic of the function.
 - `output_schema_type` (String) The output schema type of the function.
 - `output_serde_classname` (String) The output serde class name of the function.
@@ -57,6 +61,7 @@ Manages Pulsar Functions through the Functions Worker API.
 - `subscription_position` (String) The subscription position. Supported values: `Latest`, `Earliest`.
 - `timeout_ms` (Number) The timeout of the function in milliseconds.
 - `topics_pattern` (String) The input topics pattern of the function. The pattern is a regex expression. The function consumes from all topics matching the pattern.
+- `use_thread_local_producers` (Boolean) Whether to use thread local producers
 - `user_config` (Map of String) User-defined config key/values
 
 ### Read-Only

pulsar/resource_pulsar_function.go

@@ -102,6 +102,24 @@ var functionInputSourceKeys = []string{
 	resourceFunctionInputSpecsKey,
 }
 
+var functionProducerConfigKeys = []string{
+	resourceFunctionPCMaxPendingMsgKey,
+	resourceFunctionPCMaxPendingMsgAcrossPartitionKey,
+	resourceFunctionPCUseThreadLocalProducersKey,
+	resourceFunctionPCBatchBuilderKey,
+	resourceFunctionPCCompressionTypeKey,
+}
+
+// Producer configuration for the function's output topic. The attribute names mirror the ones
+// pulsar_source already exposes, so the two resources read the same way.
+const (
+	resourceFunctionPCMaxPendingMsgKey                = "max_pending_messages"
+	resourceFunctionPCMaxPendingMsgAcrossPartitionKey = "max_pending_messages_across_partitions"
+	resourceFunctionPCUseThreadLocalProducersKey      = "use_thread_local_producers"
+	resourceFunctionPCBatchBuilderKey                 = "batch_builder"
+	resourceFunctionPCCompressionTypeKey              = "compression_type"
+)
+
 const (
 	runtimeOptionSinkConfigKey         = "sinkConfig"
 	runtimeOptionSourceConfigKey       = "sourceConfig"
@@ -177,6 +195,15 @@ func init() {
 		resourceFunctionUserConfig:              "User-defined config key/values",
 		resourceFunctionSinkConfigKey:           "Sink configuration key/values serialized into custom_runtime_options.",
 		resourceFunctionSourceConfigKey:         "Source configuration key/values serialized into custom_runtime_options.",
+		//nolint:lll
+		resourceFunctionPCMaxPendingMsgKey: "The maximum size of a queue holding pending messages",
+		//nolint:lll
+		resourceFunctionPCMaxPendingMsgAcrossPartitionKey: "The maximum number of pending messages across partitions",
+		resourceFunctionPCUseThreadLocalProducersKey:      "Whether to use thread local producers",
+		//nolint:lll
+		resourceFunctionPCBatchBuilderKey: "BatchBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED.",
+		//nolint:lll
+		resourceFunctionPCCompressionTypeKey: "Set the compression type for the producer. Pulsar Functions default to LZ4. Supported compression types are: LZ4, ZLIB, ZSTD, SNAPPY and NONE",
 	}
 }
 
@@ -504,6 +531,32 @@ func resourcePulsarFunction() *schema.Resource {
 				Computed:    true,
 				Description: resourceFunctionDescriptions[resourceFunctionDiskKey],
 			},
+			resourceFunctionPCMaxPendingMsgKey: {
+				Type:        schema.TypeInt,
+				Optional:    true,
+				Description: resourceFunctionDescriptions[resourceFunctionPCMaxPendingMsgKey],
+			},
+			resourceFunctionPCMaxPendingMsgAcrossPartitionKey: {
+				Type:        schema.TypeInt,
+				Optional:    true,
+				Description: resourceFunctionDescriptions[resourceFunctionPCMaxPendingMsgAcrossPartitionKey],
+			},
+			resourceFunctionPCUseThreadLocalProducersKey: {
+				Type:        schema.TypeBool,
+				Optional:    true,
+				Description: resourceFunctionDescriptions[resourceFunctionPCUseThreadLocalProducersKey],
+			},
+			resourceFunctionPCBatchBuilderKey: {
+				Type:        schema.TypeString,
+				Optional:    true,
+				Description: resourceFunctionDescriptions[resourceFunctionPCBatchBuilderKey],
+			},
+			resourceFunctionPCCompressionTypeKey: {
+				Type:        schema.TypeString,
+				Optional:    true,
+				Computed:    true,
+				Description: resourceFunctionDescriptions[resourceFunctionPCCompressionTypeKey],
+			},
 			resourceFunctionUserConfig: {
 				Type:        schema.TypeMap,
 				Optional:    true,
@@ -1204,6 +1257,8 @@ func marshalFunctionConfig(d *schema.ResourceData) (*utils.FunctionConfig, error
 		functionConfig.UserConfig = interMap
 	}
 
+	functionConfig.ProducerConfig = marshalFunctionProducerConfig(d)
+
 	return functionConfig, nil
 }
 
@@ -1297,7 +1352,91 @@ func flattenFunctionInputSpec(topic string, consumerConfig utils.ConsumerConfig)
 	return spec
 }
 
+// marshalFunctionProducerConfig builds the output producer's configuration, mirroring how
+// pulsar_source populates the same struct. It returns nil when nothing is configured so the
+// request is unchanged for functions that do not set any of these.
+func marshalFunctionProducerConfig(d *schema.ResourceData) *utils.ProducerConfig {
+	producerConfig := &utils.ProducerConfig{}
+	configured := false
+
+	if inter, ok := d.GetOk(resourceFunctionPCMaxPendingMsgKey); ok {
+		producerConfig.MaxPendingMessages = inter.(int)
+		configured = true
+	}
+
+	if inter, ok := d.GetOk(resourceFunctionPCMaxPendingMsgAcrossPartitionKey); ok {
+		producerConfig.MaxPendingMessagesAcrossPartitions = inter.(int)
+		configured = true
+	}
+
+	if inter, ok := d.GetOk(resourceFunctionPCUseThreadLocalProducersKey); ok {
+		producerConfig.UseThreadLocalProducers = inter.(bool)
+		configured = true
+	}
+
+	if inter, ok := d.GetOk(resourceFunctionPCBatchBuilderKey); ok {
+		producerConfig.BatchBuilder = inter.(string)
+		configured = true
+	}
+
+	if inter, ok := d.GetOk(resourceFunctionPCCompressionTypeKey); ok {
+		producerConfig.CompressionType = inter.(string)
+		configured = true
+	}
+
+	if !configured && d.Id() != "" && d.HasChanges(functionProducerConfigKeys...) {
+		// FunctionConfigUtils.validateUpdate() preserves the existing producer config when the
+		// request field is nil. Send the Function default explicitly when the last configured
+		// producer attribute is removed so zero-valued settings are actually cleared.
+		return &utils.ProducerConfig{CompressionType: "LZ4"}
+	}
+
+	if !configured {
+		return nil
+	}
+
+	return producerConfig
+}
+
+// unmarshalFunctionProducerConfig writes the complete output producer configuration into state.
+// Zero values must be written too: skipping them leaves an earlier non-zero state value behind when
+// the producer configuration is removed or changed outside Terraform.
+func unmarshalFunctionProducerConfig(functionConfig utils.FunctionConfig, d *schema.ResourceData) error {
+	producerConfig := functionConfig.ProducerConfig
+	if producerConfig == nil {
+		producerConfig = &utils.ProducerConfig{}
+	}
+
+	if err := d.Set(resourceFunctionPCMaxPendingMsgKey, producerConfig.MaxPendingMessages); err != nil {
+		return err
+	}
+
+	if err := d.Set(resourceFunctionPCMaxPendingMsgAcrossPartitionKey,
+		producerConfig.MaxPendingMessagesAcrossPartitions); err != nil {
+		return err
+	}
+
+	if err := d.Set(resourceFunctionPCUseThreadLocalProducersKey,
+		producerConfig.UseThreadLocalProducers); err != nil {
+		return err
+	}
+
+	if err := d.Set(resourceFunctionPCBatchBuilderKey, producerConfig.BatchBuilder); err != nil {
+		return err
+	}
+
+	if err := d.Set(resourceFunctionPCCompressionTypeKey, producerConfig.CompressionType); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func unmarshalFunctionConfig(functionConfig utils.FunctionConfig, d *schema.ResourceData) error {
+	if err := unmarshalFunctionProducerConfig(functionConfig, d); err != nil {
+		return err
+	}
+
 	if functionConfig.Jar != nil {
 		err := d.Set(resourceFunctionJarKey, *functionConfig.Jar)
 		if err != nil {

pulsar/resource_pulsar_function_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func init() {
@@ -106,6 +107,14 @@ func TestFunction(t *testing.T) {
 						config.InputSpecs["public/default/schema-input"].ReceiverQueueSize)
 					assert.Equal(t, "STRING", config.InputSpecs["public/default/schema-input"].SchemaType)
 
+					// #220 part A: the output producer configuration must round-trip.
+					require.NotNil(t, config.ProducerConfig)
+					assert.Equal(t, "ZSTD", config.ProducerConfig.CompressionType)
+					assert.Equal(t, "KEY_BASED", config.ProducerConfig.BatchBuilder)
+					assert.Equal(t, 1000, config.ProducerConfig.MaxPendingMessages)
+					assert.Equal(t, 50000, config.ProducerConfig.MaxPendingMessagesAcrossPartitions)
+					assert.True(t, config.ProducerConfig.UseThreadLocalProducers)
+
 					return nil
 				}),
 			},
@@ -147,6 +156,13 @@ func TestFunction(t *testing.T) {
 					// have reset it to a default ConsumerConfig here.
 					assert.Equal(t, "avro", config.InputSpecs["public/default/input1"].SchemaType)
 
+					require.NotNil(t, config.ProducerConfig)
+					assert.Equal(t, "LZ4", config.ProducerConfig.CompressionType)
+					assert.Empty(t, config.ProducerConfig.BatchBuilder)
+					assert.Zero(t, config.ProducerConfig.MaxPendingMessages)
+					assert.Zero(t, config.ProducerConfig.MaxPendingMessagesAcrossPartitions)
+					assert.False(t, config.ProducerConfig.UseThreadLocalProducers)
+
 					return nil
 				}),
 			},

pulsar/resource_pulsar_function_unit_test.go

@@ -13,6 +13,26 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestUnmarshalFunctionProducerConfigClearsZeroValues(t *testing.T) {
+	d := schema.TestResourceDataRaw(t, resourcePulsarFunction().Schema, map[string]interface{}{
+		resourceFunctionPCMaxPendingMsgKey:                1000,
+		resourceFunctionPCMaxPendingMsgAcrossPartitionKey: 50000,
+		resourceFunctionPCUseThreadLocalProducersKey:      true,
+		resourceFunctionPCBatchBuilderKey:                 "KEY_BASED",
+		resourceFunctionPCCompressionTypeKey:              "ZSTD",
+	})
+
+	err := unmarshalFunctionProducerConfig(utils.FunctionConfig{
+		ProducerConfig: &utils.ProducerConfig{},
+	}, d)
+	require.NoError(t, err)
+	assert.Zero(t, d.Get(resourceFunctionPCMaxPendingMsgKey))
+	assert.Zero(t, d.Get(resourceFunctionPCMaxPendingMsgAcrossPartitionKey))
+	assert.False(t, d.Get(resourceFunctionPCUseThreadLocalProducersKey).(bool))
+	assert.Empty(t, d.Get(resourceFunctionPCBatchBuilderKey))
+	assert.Empty(t, d.Get(resourceFunctionPCCompressionTypeKey))
+}
+
 func TestMergeFunctionCustomRuntimeOptions(t *testing.T) {
 	base := `{"foo":"bar","sinkConfig":{"old":"value"},"sourceConfig":{"keep":"me"}}`
 	sinkConfig := map[string]interface{}{

pulsar/testdata/function/main.tf

@@ -73,6 +73,13 @@ resource "pulsar_function" "function-1" {
   log_topic                       = "public/default/lt"
   timeout_ms                      = 6666
 
+  # Output producer configuration (#220 part A).
+  compression_type                       = "ZSTD"
+  batch_builder                          = "KEY_BASED"
+  max_pending_messages                   = 1000
+  max_pending_messages_across_partitions = 50000
+  use_thread_local_producers             = true
+
   custom_runtime_options = jsonencode(
     {
       "env" : {

pulsar/testdata/function/main_updated.tf

@@ -71,6 +71,9 @@ resource "pulsar_function" "function-1" {
   log_topic                       = "public/default/lt"
   timeout_ms                      = 6666
 
+  # Reset the output producer to the Function default while removing the other producer settings.
+  compression_type = "LZ4"
+
   custom_runtime_options = jsonencode(
     {
       "env" : {

pulsar/testdata/function/main_zero_queue.tf

@@ -68,6 +68,8 @@ resource "pulsar_function" "function-1" {
   log_topic                       = "public/default/lt"
   timeout_ms                      = 6666
 
+  compression_type = "LZ4"
+
   custom_runtime_options = jsonencode(
     {
       "env" : {