From 73fe688a439c3ab9b56f2d249f16505292391f4b Mon Sep 17 00:00:00 2001
From: Ethan-Merrill <92824266+Ethan-Merrill@users.noreply.github.com>
Date: Tue, 18 Jul 2023 20:37:03 -0600
Subject: [PATCH] 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 | 4 ++++
 charts/pulsar/templates/broker-configmap.yaml     | 3 +++
 charts/pulsar/values.yaml                         | 2 ++
 3 files changed, 9 insertions(+)

diff --git a/charts/pulsar/templates/bookkeeper-configmap.yaml b/charts/pulsar/templates/bookkeeper-configmap.yaml
index 8ed5f9b..f133f1f 100644
--- a/charts/pulsar/templates/bookkeeper-configmap.yaml
+++ b/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 }}
diff --git a/charts/pulsar/templates/broker-configmap.yaml b/charts/pulsar/templates/broker-configmap.yaml
index b5278c9..c02a4df 100644
--- a/charts/pulsar/templates/broker-configmap.yaml
+++ b/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
diff --git a/charts/pulsar/values.yaml b/charts/pulsar/values.yaml
index f347d2f..34c2898 100644
--- a/charts/pulsar/values.yaml
+++ b/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