[improve][fn] Introduce NewOutputMessageWithError to enable error handling (#24122)

Co-authored-by: Simon Beeli <[email protected]>
Co-authored-by: Zixuan Liu <[email protected]>
(cherry picked from commit d8e2743)

Author: 3 people

pulsar-function-go/pf/context.go

@@ -34,12 +34,13 @@ import (
 // message, what are our operating constraints, etc can be accessed by the
 // executing function
 type FunctionContext struct {
-	instanceConf  *instanceConf
-	userConfigs   map[string]interface{}
-	logAppender   *LogAppender
-	outputMessage func(topic string) pulsar.Producer
-	userMetrics   sync.Map
-	record        pulsar.Message
+	instanceConf           *instanceConf
+	userConfigs            map[string]interface{}
+	logAppender            *LogAppender
+	outputMessage          func(topic string) pulsar.Producer
+	outputMessageWithError func(topic string) (pulsar.Producer, error)
+	userMetrics            sync.Map
+	record                 pulsar.Message
 }
 
 // NewFuncContext returns a new Function context
@@ -161,6 +162,13 @@ func (c *FunctionContext) NewOutputMessage(topicName string) pulsar.Producer {
 	return c.outputMessage(topicName)
 }
 
+// NewOutputMessageWithError send message to the topic and returns a potential error
+// @param topicName: The name of the topic for output message
+// @return A Pulsar producer for the given topic and an error, if any.
+func (c *FunctionContext) NewOutputMessageWithError(topicName string) (pulsar.Producer, error) {
+	return c.outputMessageWithError(topicName)
+}
+
 // SetCurrentRecord sets the current message into the function context called
 // for each message before executing a handler function
 func (c *FunctionContext) SetCurrentRecord(record pulsar.Message) {

pulsar-function-go/pf/context_test.go

@@ -21,6 +21,8 @@ package pf
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"testing"
 	"time"
 
@@ -82,3 +84,46 @@ func TestFunctionContext_NewOutputMessage(t *testing.T) {
 	actualProducer := fc.NewOutputMessage(publishTopic)
 	assert.IsType(t, &MockPulsarProducer{}, actualProducer)
 }
+
+func TestFunctionContext_NewOutputMessageWithError(t *testing.T) {
+	testErr := errors.New("test error")
+
+	testCases := []struct {
+		name                 string
+		outputFunc           func(topic string) (pulsar.Producer, error)
+		expectedError        error
+		expectedProducerType *MockPulsarProducer
+	}{
+
+		{
+			name:                 "Test producer",
+			outputFunc:           func(topic string) (pulsar.Producer, error) { return &MockPulsarProducer{}, nil },
+			expectedError:        nil,
+			expectedProducerType: &MockPulsarProducer{},
+		},
+		{
+			name:                 "Test error",
+			outputFunc:           func(topic string) (pulsar.Producer, error) { return nil, errors.New("test error") },
+			expectedError:        testErr,
+			expectedProducerType: nil,
+		},
+	}
+
+	for i, testCase := range testCases {
+		t.Run(fmt.Sprintf("testCase[%d] %s", i, testCase.name), func(t *testing.T) {
+
+			fc := NewFuncContext()
+			publishTopic := "publish-topic"
+
+			fc.outputMessageWithError = testCase.outputFunc
+
+			actualProducer, err := fc.NewOutputMessageWithError(publishTopic)
+			if testCase.expectedProducerType == nil {
+				assert.Nil(t, actualProducer)
+			} else {
+				assert.IsType(t, testCase.expectedProducerType, actualProducer)
+			}
+			assert.Equal(t, testCase.expectedError, err)
+		})
+	}
+}

pulsar-function-go/pf/instance.go

@@ -77,6 +77,15 @@ func newGoInstance() *goInstance {
 		return producer
 	}
 
+	goInstance.context.outputMessageWithError = func(topic string) (pulsar.Producer, error) {
+		producer, err := goInstance.getProducer(topic)
+		if err != nil {
+			log.Errorf("getting producer failed, error is:%v", err)
+			return nil, err
+		}
+		return producer, nil
+	}
+
 	goInstance.lastHealthCheckTS = now.UnixNano()
 	goInstance.properties = make(map[string]string)
 	goInstance.stats = NewStatWithLabelValues(goInstance.getMetricsLabels()...)

pulsar-function-go/pf/util_test.go

@@ -44,8 +44,8 @@ func TestUtils(t *testing.T) {
 	expectedFQFN := getDefaultSubscriptionName(tenant, namespace, name)
 	assert.Equal(t, expectedFQFN, fqfn)
 
-	actualtMap := getProperties(fqfn, 100)
-	assert.Equal(t, propertiesMap, actualtMap)
+	actualMap := getProperties(fqfn, 100)
+	assert.Equal(t, propertiesMap, actualMap)
 
 	expectedRes := getFullyQualifiedInstanceID(tenant, namespace, name, instanceID)
 	assert.Equal(t, expectedRes, "pulsar/function/go:100")