#325 add State Store configs to support stateful functions in Pulsar (#359)

* add State Store configs

* more comment

* fix CI

* fix style

* stash

* add tests

* add stateful sample and integration tests

* fix tests

* more output

* add test step

* address comments

Author: freeznet

.ci/clusters/compute_v1alpha1_function_stateful.yaml

@@ -0,0 +1,44 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: java-function-stateful-sample
+  namespace: default
+spec:
+  image: streamnative/pulsar-functions-java-sample:2.10.0.0-rc10
+  className: org.apache.pulsar.functions.api.examples.WordCountFunction
+  forwardSourceMessageProperty: true
+  maxPendingAsyncRequests: 1000
+  replicas: 1
+  maxReplicas: 5
+  logTopic: persistent://public/default/logging-function-logs
+  input:
+    topics:
+    - persistent://public/default/java-function-stateful-input-topic
+    typeClassName: java.lang.String
+  output:
+    topic: persistent://public/default/java-function-stateful-output-topic
+    typeClassName: java.lang.String
+  resources:
+    requests:
+      cpu: "0.1"
+      memory: 1G
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+  pulsar:
+    pulsarConfig: "test-pulsar"
+  java:
+    jar: /pulsar/examples/api-examples.jar
+  clusterName: test-pulsar
+  autoAck: true
+  statefulConfig:
+    pulsar:
+      serviceUrl: "bk://sn-platform-pulsar-bk.default.svc.cluster.local:4181"
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-pulsar
+data:
+    webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+    brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650

.ci/clusters/values.yaml

@@ -112,3 +112,6 @@ broker:
 
 proxy:
   replicaCount: 1
+
+functions:
+  functionState: true

.ci/verify_function_mesh.sh

@@ -73,4 +73,12 @@ case ${1} in
     ci::verify_java_function function-builtin-hpa-sample
     ci::verify_hpa function-builtin-hpa-sample
     ;;
+  compute_v1alpha1_function_stateful)
+    ci::verify_function_mesh java-function-stateful-sample
+    sleep 60
+    ci::print_function_log java-function-stateful-sample
+    ci::verify_java_function java-function-stateful-sample
+    sleep 60
+    ci::print_function_log java-function-stateful-sample
+    ;;
 esac

.github/workflows/test-integration-kind-samples.yml

@@ -133,6 +133,16 @@ jobs:
           .ci/verify_function_mesh.sh compute_v1alpha1_function_builtin_hpa
           kubectl delete -f .ci/clusters/compute_v1alpha1_function_builtin_hpa.yaml
 
+      - name: Test Java Stateful Function
+        run: |
+          kubectl apply -f .ci/clusters/compute_v1alpha1_function_stateful.yaml
+          kubectl get all
+
+      - name: Verify Java Stateful Function
+        run: |
+          .ci/verify_function_mesh.sh compute_v1alpha1_function_stateful
+          kubectl delete -f .ci/clusters/compute_v1alpha1_function_stateful.yaml
+
       - name: Setup tmate session
         uses: mxschmitt/action-tmate@v3
         if: failure()

api/v1alpha1/common.go

@@ -35,6 +35,10 @@ type Messaging struct {
 	Pulsar *PulsarMessaging `json:"pulsar,omitempty"`
 }
 
+type Stateful struct {
+	Pulsar *PulsarStateStore `json:"pulsar,omitempty"`
+}
+
 type PulsarMessaging struct {
 	// The config map need to contain the following fields
 	// webServiceURL
@@ -54,6 +58,25 @@ type PulsarMessaging struct {
 	TLSSecret string `json:"tlsSecret,omitempty"`
 }
 
+type PulsarStateStore struct {
+	// The service url points to the state store service
+	// By default, the state store service is bookkeeper table service
+	ServiceURL string `json:"serviceUrl"`
+
+	// The state store config for Java runtime
+	JavaProvider *PulsarStateStoreJavaProvider `json:"javaProvider,omitempty"`
+}
+
+type PulsarStateStoreJavaProvider struct {
+	// The java class name of the state store provider implementation
+	// The class must implement `org.apache.pulsar.functions.instance.state.StateStoreProvider` interface
+	// If not set, `org.apache.pulsar.functions.instance.state.BKStateStoreProviderImpl` will be used
+	ClassName string `json:"className"`
+
+	// The configmap of the configuration for the state store provider
+	Config *Config `json:"config,omitempty"`
+}
+
 type PodPolicy struct {
 	// Labels specifies the labels to attach to pod the operator creates for the cluster.
 	Labels map[string]string `json:"labels,omitempty"`

api/v1alpha1/function_types.go

@@ -85,6 +85,9 @@ type FunctionSpec struct {
 
 	// Image pull policy, one of Always, Never, IfNotPresent, default to IfNotPresent.
 	ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy,omitempty"`
+
+	// +kubebuilder:validation:Optional
+	StateConfig *Stateful `json:"statefulConfig,omitempty"`
 }
 
 // FunctionStatus defines the observed state of Function

api/v1alpha1/function_webhook.go

@@ -217,6 +217,11 @@ func (r *Function) ValidateCreate() error {
 		allErrs = append(allErrs, fieldErr)
 	}
 
+	fieldErr = validateStatefulFunctionConfigs(r.Spec.StateConfig, r.Spec.Runtime)
+	if fieldErr != nil {
+		allErrs = append(allErrs, fieldErr)
+	}
+
 	if len(allErrs) == 0 {
 		return nil
 	}

api/v1alpha1/validate.go

@@ -34,7 +34,8 @@ func validateJavaRuntime(java *JavaRuntime, className string) []*field.Error {
 			allErrs = append(allErrs, e)
 		}
 		if java.Jar == "" {
-			e := field.Invalid(field.NewPath("spec").Child("java", "jar"), java.Jar, "jar cannot be empty in java runtime")
+			e := field.Invalid(field.NewPath("spec").Child("java", "jar"), java.Jar,
+				"jar cannot be empty in java runtime")
 			allErrs = append(allErrs, e)
 		}
 		if java.JarLocation != "" {
@@ -56,7 +57,8 @@ func validatePythonRuntime(python *PythonRuntime, className string) []*field.Err
 			allErrs = append(allErrs, e)
 		}
 		if python.Py == "" {
-			e := field.Invalid(field.NewPath("spec").Child("python", "py"), python.Py, "py cannot be empty in python runtime")
+			e := field.Invalid(field.NewPath("spec").Child("python", "py"), python.Py,
+				"py cannot be empty in python runtime")
 			allErrs = append(allErrs, e)
 		}
 		if python.PyLocation != "" {
@@ -74,7 +76,8 @@ func validateGolangRuntime(golang *GoRuntime) []*field.Error {
 	var allErrs field.ErrorList
 	if golang != nil {
 		if golang.Go == "" {
-			e := field.Invalid(field.NewPath("spec").Child("golang", "go"), golang.Go, "go cannot be empty in golang runtime")
+			e := field.Invalid(field.NewPath("spec").Child("golang", "go"), golang.Go,
+				"go cannot be empty in golang runtime")
 			allErrs = append(allErrs, e)
 		}
 		if golang.GoLocation != "" {
@@ -102,7 +105,8 @@ func validateReplicasAndMaxReplicas(replicas, maxReplicas *int32) []*field.Error
 	}
 
 	if maxReplicas != nil && replicas != nil && *replicas > *maxReplicas {
-		e := field.Invalid(field.NewPath("spec").Child("maxReplicas"), *maxReplicas, "maxReplicas must be greater than or equal to replicas")
+		e := field.Invalid(field.NewPath("spec").Child("maxReplicas"), *maxReplicas,
+			"maxReplicas must be greater than or equal to replicas")
 		allErrs = append(allErrs, e)
 	}
 	return allErrs
@@ -117,38 +121,45 @@ func validateResourceRequirement(requirements corev1.ResourceRequirements) *fiel
 
 func validateTimeout(timeout int32, processingGuarantee ProcessGuarantee) *field.Error {
 	if timeout != 0 && processingGuarantee == EffectivelyOnce {
-		return field.Invalid(field.NewPath("spec").Child("timeout"), timeout, "message timeout can only be set for AtleastOnce processing guarantee")
+		return field.Invalid(field.NewPath("spec").Child("timeout"), timeout,
+			"message timeout can only be set for AtleastOnce processing guarantee")
 	}
 	return nil
 }
 
-func validateMaxMessageRetry(maxMessageRetry int32, processingGuarantee ProcessGuarantee, deadLetterTopic string) []*field.Error {
+func validateMaxMessageRetry(maxMessageRetry int32, processingGuarantee ProcessGuarantee,
+	deadLetterTopic string) []*field.Error {
 	var allErrs field.ErrorList
 	if maxMessageRetry > 0 && processingGuarantee == EffectivelyOnce {
-		e := field.Invalid(field.NewPath("spec").Child("maxMessageRetry"), maxMessageRetry, "MaxMessageRetries and Effectively once are not compatible")
+		e := field.Invalid(field.NewPath("spec").Child("maxMessageRetry"), maxMessageRetry,
+			"MaxMessageRetries and Effectively once are not compatible")
 		allErrs = append(allErrs, e)
 	}
 
 	if maxMessageRetry <= 0 && deadLetterTopic != "" {
-		e := field.Invalid(field.NewPath("spec").Child("maxMessageRetry"), maxMessageRetry, "dead letter topic is set but max message retry is set to infinity")
+		e := field.Invalid(field.NewPath("spec").Child("maxMessageRetry"), maxMessageRetry,
+			"dead letter topic is set but max message retry is set to infinity")
 		allErrs = append(allErrs, e)
 	}
 	return allErrs
 }
 
 func validateRetainKeyOrdering(retainKeyOrdering bool, processingGuarantee ProcessGuarantee) *field.Error {
 	if retainKeyOrdering && processingGuarantee == EffectivelyOnce {
-		return field.Invalid(field.NewPath("spec").Child("retainKeyOrdering"), retainKeyOrdering, "when effectively once processing guarantee is specified, retain Key ordering cannot be set")
+		return field.Invalid(field.NewPath("spec").Child("retainKeyOrdering"), retainKeyOrdering,
+			"when effectively once processing guarantee is specified, retain Key ordering cannot be set")
 	}
 	return nil
 }
 
 func validateRetainOrderingConflicts(retainKeyOrdering bool, retainOrdering bool) []*field.Error {
 	var allErrs field.ErrorList
 	if retainKeyOrdering && retainOrdering {
-		e := field.Invalid(field.NewPath("spec").Child("retainKeyOrdering"), retainKeyOrdering, "only one of retain ordering or retain key ordering can be set")
+		e := field.Invalid(field.NewPath("spec").Child("retainKeyOrdering"), retainKeyOrdering,
+			"only one of retain ordering or retain key ordering can be set")
 		allErrs = append(allErrs, e)
-		e = field.Invalid(field.NewPath("spec").Child("retainOrdering"), retainOrdering, "only one of retain ordering or retain key ordering can be set")
+		e = field.Invalid(field.NewPath("spec").Child("retainOrdering"), retainOrdering,
+			"only one of retain ordering or retain key ordering can be set")
 		allErrs = append(allErrs, e)
 	}
 	return allErrs
@@ -158,7 +169,8 @@ func validateFunctionConfig(config *Config) *field.Error {
 	if config != nil {
 		_, err := config.MarshalJSON()
 		if err != nil {
-			return field.Invalid(field.NewPath("spec").Child("funcConfig"), config, "function config is invalid: "+err.Error())
+			return field.Invalid(field.NewPath("spec").Child("funcConfig"), config,
+				"function config is invalid: "+err.Error())
 		}
 	}
 	return nil
@@ -168,7 +180,8 @@ func validateSinkConfig(config *Config) *field.Error {
 	if config != nil {
 		_, err := config.MarshalJSON()
 		if err != nil {
-			return field.Invalid(field.NewPath("spec").Child("sinkConfig"), config, "sink config is invalid: "+err.Error())
+			return field.Invalid(field.NewPath("spec").Child("sinkConfig"), config,
+				"sink config is invalid: "+err.Error())
 		}
 	}
 	return nil
@@ -178,7 +191,8 @@ func validateSourceConfig(config *Config) *field.Error {
 	if config != nil {
 		_, err := config.MarshalJSON()
 		if err != nil {
-			return field.Invalid(field.NewPath("spec").Child("sourceConfig"), config, "source config is invalid: "+err.Error())
+			return field.Invalid(field.NewPath("spec").Child("sourceConfig"), config,
+				"source config is invalid: "+err.Error())
 		}
 	}
 	return nil
@@ -188,7 +202,8 @@ func validateSecretsMap(secrets map[string]SecretRef) *field.Error {
 	if secrets != nil {
 		_, err := json.Marshal(secrets)
 		if err != nil {
-			return field.Invalid(field.NewPath("spec").Child("secretsMap"), secrets, "secrets map is invalid: "+err.Error())
+			return field.Invalid(field.NewPath("spec").Child("secretsMap"), secrets,
+				"secrets map is invalid: "+err.Error())
 		}
 	}
 	return nil
@@ -247,14 +262,18 @@ func validateInputOutput(input *InputConf, output *OutputConf) []*field.Error {
 			}
 			if output.ProducerConf != nil && output.ProducerConf.CryptoConfig != nil {
 				if output.ProducerConf.CryptoConfig.CryptoKeyReaderClassName == "" {
-					e := field.Invalid(field.NewPath("spec").Child("output", "producerConf", "cryptoConfig", "cryptoKeyReaderClassName"),
-						output.ProducerConf.CryptoConfig.CryptoKeyReaderClassName, "cryptoKeyReader class name required")
+					e := field.Invalid(field.NewPath("spec").Child("output", "producerConf", "cryptoConfig",
+						"cryptoKeyReaderClassName"),
+						output.ProducerConf.CryptoConfig.CryptoKeyReaderClassName,
+						"cryptoKeyReader class name required")
 					allErrs = append(allErrs, e)
 				}
 
 				if len(output.ProducerConf.CryptoConfig.EncryptionKeys) == 0 {
-					e := field.Invalid(field.NewPath("spec").Child("output", "producerConf", "cryptoConfig", "encryptionKeys"),
-						output.ProducerConf.CryptoConfig.EncryptionKeys, "must provide encryption key name for crypto key reader")
+					e := field.Invalid(field.NewPath("spec").Child("output", "producerConf", "cryptoConfig",
+						"encryptionKeys"),
+						output.ProducerConf.CryptoConfig.EncryptionKeys,
+						"must provide encryption key name for crypto key reader")
 					allErrs = append(allErrs, e)
 				}
 			}
@@ -268,7 +287,8 @@ func validateLogTopic(logTopic string) *field.Error {
 	if logTopic != "" {
 		err := isValidTopicName(logTopic)
 		if err != nil {
-			return field.Invalid(field.NewPath("spec").Child("logTopic"), logTopic, fmt.Sprintf("Log topic %s is invalid", logTopic))
+			return field.Invalid(field.NewPath("spec").Child("logTopic"), logTopic,
+				fmt.Sprintf("Log topic %s is invalid", logTopic))
 		}
 	}
 	return nil
@@ -278,7 +298,8 @@ func validateDeadLetterTopic(deadLetterTopic string) *field.Error {
 	if deadLetterTopic != "" {
 		err := isValidTopicName(deadLetterTopic)
 		if err != nil {
-			return field.Invalid(field.NewPath("spec").Child("deadLetterTopic"), deadLetterTopic, fmt.Sprintf("DeadLetter topic %s is invalid", deadLetterTopic))
+			return field.Invalid(field.NewPath("spec").Child("deadLetterTopic"), deadLetterTopic,
+				fmt.Sprintf("DeadLetter topic %s is invalid", deadLetterTopic))
 		}
 	}
 	return nil
@@ -290,3 +311,23 @@ func validateAutoAck(autoAck *bool) *field.Error {
 	}
 	return nil
 }
+
+func validateStatefulFunctionConfigs(statefulFunctionConfigs *Stateful, runtime Runtime) *field.Error {
+	if statefulFunctionConfigs != nil {
+		if statefulFunctionConfigs.Pulsar != nil {
+			if isGolangRuntime(runtime) {
+				return field.Invalid(field.NewPath("spec").Child("statefulConfig"), runtime.Golang,
+					"Golang function do not support stateful function yet")
+			}
+			if statefulFunctionConfigs.Pulsar.ServiceURL == "" {
+				return field.Invalid(field.NewPath("spec").Child("statefulConfig", "pulsar", "serviceUrl"),
+					statefulFunctionConfigs.Pulsar.ServiceURL, "serviceUrl cannot be empty")
+			}
+		}
+	}
+	return nil
+}
+
+func isGolangRuntime(runtime Runtime) bool {
+	return runtime.Golang != nil && runtime.Python == nil && runtime.Java == nil
+}