Add support for stateful functions using the bookie as state storage (#171)

### Motivation

Enables support for using the Pulsar bookies as persistent state storage for functions.

### Modifications

- Added an option to enable/disable using bookies as state storage
- Adds extra server components options to the bookkeeper to enable necessary features for bookies to be used as state storage
- Adds stateStorageServiceUrl to the broker configmap

charts/pulsar/templates/bookkeeper-configmap.yaml

@@ -55,6 +55,10 @@ data:
   {{- else }}
   ledgerDirectories: "/pulsar/data/bookkeeper/ledgers"
   {{- end }}
+  {{- if .Values.functions.useBookieAsStateStore }}
+  # Stateful function config
+  extraServerComponents: "org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent"
+  {{- end }}
   # TLS config
   {{- include "pulsar.bookkeeper.config.tls" . | nindent 2 }}
 {{ toYaml .Values.bookkeeper.configData | indent 2 }}

charts/pulsar/templates/broker-configmap.yaml

@@ -50,6 +50,9 @@ data:
   {{- end }}
   {{- if or .Values.components.functions .Values.extra.functionsAsPods }}
   functionsWorkerEnabled: "true"
+  {{- if .Values.functions.useBookieAsStateStore }}
+  PF_stateStorageServiceUrl: "bk://{{ template "pulsar.fullname" . }}-{{ .Values.bookkeeper.component }}:{{ .Values.bookkeeper.ports.statestore }}"
+  {{- end }}
   PF_functionRuntimeFactoryClassName: "org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactory"
   PF_pulsarFunctionsCluster: {{ template "pulsar.cluster.name" . }}
   PF_connectorsDirectory: ./connectors

charts/pulsar/values.yaml

@@ -437,6 +437,7 @@ bookkeeper:
   ports:
     http: 8000
     bookie: 3181
+    statestore: 4181
   # nodeSelector:
     # cloud.google.com/gke-nodepool: default-pool
   probe:
@@ -796,6 +797,7 @@ broker:
 ##
 functions:
   component: functions-worker
+  useBookieAsStateStore: false
   ## Pulsar: Functions Worker ClusterRole or Role
   ## templates/broker-rbac.yaml
   # Default is false which deploys functions with ClusterRole and ClusterRoleBinding at the cluster level