Use PIP-45 metadata store config to replace deprecated ZK config and make PulsarMetadataBookieDriver configurable in BK (#576)

.ci/chart_test.sh

@@ -62,6 +62,8 @@ if [[ "$UPGRADE_FROM_VERSION" != "" ]]; then
     install_type="upgrade"
     echo "Wait 10 seconds"
     sleep 10
+    # check pulsar environment
+    ci::check_pulsar_environment
     # test that we can access the admin api
     ci::test_pulsar_admin_api_access
     # produce messages with old version of pulsar and consume with new version

.ci/configure_ci_runner_for_debugging.sh

@@ -27,7 +27,7 @@ function k9s() {
     # install k9s on the fly
     if [ ! -x /usr/local/bin/k9s ]; then
         echo "Installing k9s..."
-        curl -L -s https://github.com/derailed/k9s/releases/download/v0.32.5/k9s_Linux_amd64.tar.gz | sudo tar xz -C /usr/local/bin k9s
+        curl -L -s https://github.com/derailed/k9s/releases/download/v0.40.5/k9s_Linux_amd64.tar.gz | sudo tar xz -C /usr/local/bin k9s
     fi
     command k9s "$@"
 }

.ci/values-common.yaml

@@ -55,6 +55,12 @@ bookkeeper:
     diskUsageWarnThreshold: "0.999"
     PULSAR_PREFIX_diskUsageThreshold: "0.999"
     PULSAR_PREFIX_diskUsageWarnThreshold: "0.999"
+    # minimal memory use for bookkeeper
+    # https://bookkeeper.apache.org/docs/reference/config#db-ledger-storage-settings
+    dbStorage_writeCacheMaxSizeMb: "32"
+    dbStorage_readAheadCacheMaxSizeMb: "32"
+    dbStorage_rocksDB_writeBufferSizeMB: "8"
+    dbStorage_rocksDB_blockCacheSize: "8388608"    
 
 broker:
   replicaCount: 1

charts/pulsar/templates/_bookkeeper.tpl

@@ -98,11 +98,22 @@ Define bookie common config
 */}}
 {{- define "pulsar.bookkeeper.config.common" -}}
 {{- if .Values.components.zookeeper }}
+{{- if (and (hasKey .Values.pulsar_metadata "bookkeeper") .Values.pulsar_metadata.bookkeeper.usePulsarMetadataBookieDriver) }}
+# there's a bug when using PulsarMetadataBookieDriver since it always appends /ledgers to the metadataServiceUri
+# Possibly a bug in org.apache.pulsar.metadata.bookkeeper.AbstractMetadataDriver#resolveLedgersRootPath in Pulsar code base
+metadataServiceUri: "metadata-store:zk:{{ template "pulsar.zookeeper.connect" . }}{{ .Values.metadataPrefix }}"
+{{- else }}
 zkServers: "{{ template "pulsar.zookeeper.connect" . }}"
 zkLedgersRootPath: "{{ .Values.metadataPrefix }}/ledgers"
+{{- end }}
 {{- else if .Values.components.oxia }}
 metadataServiceUri: "{{ template "pulsar.oxia.metadata.url.bookkeeper" . }}"
 {{- end }}
+{{- /* metadataStoreSessionTimeoutMillis maps to zkTimeout in bookkeeper.conf for both zookeeper and oxia metadata stores */}}
+{{- if (and (hasKey .Values.pulsar_metadata "bookkeeper") (hasKey .Values.pulsar_metadata.bookkeeper "metadataStoreSessionTimeoutMillis")) }}
+zkTimeout: "{{ .Values.pulsar_metadata.bookkeeper.metadataStoreSessionTimeoutMillis }}"
+{{- end }}
+
 # enable bookkeeper http server
 httpServerEnabled: "true"
 httpServerPort: "{{ .Values.bookkeeper.ports.http }}"

charts/pulsar/templates/broker-configmap.yaml

@@ -29,12 +29,19 @@ metadata:
 data:
   # Metadata settings
   {{- if .Values.components.zookeeper }}
-  zookeeperServers: "{{ template "pulsar.zookeeper.connect" . }}{{ .Values.metadataPrefix }}"
+  metadataStoreUrl: "zk:{{ template "pulsar.zookeeper.connect" . }}{{ .Values.metadataPrefix }}"
+  {{- $configMetadataStoreUrl := "" }}
   {{- if .Values.pulsar_metadata.configurationStore }}
-  configurationStoreServers: "{{ template "pulsar.configurationStore.connect" . }}{{ .Values.pulsar_metadata.configurationStoreMetadataPrefix }}"
+  {{- $configMetadataStoreUrl = printf "zk:%s%s" (include "pulsar.configurationStore.connect" .) .Values.pulsar_metadata.configurationStoreMetadataPrefix }}
+  {{- else }}
+  {{- $configMetadataStoreUrl = printf "zk:%s%s" (include "pulsar.zookeeper.connect" .) .Values.metadataPrefix }}
   {{- end }}
-  {{- if not .Values.pulsar_metadata.configurationStore }}
-  configurationStoreServers: "{{ template "pulsar.zookeeper.connect" . }}{{ .Values.metadataPrefix }}"
+  configurationMetadataStoreUrl: "{{ $configMetadataStoreUrl }}"
+  # setting bookkeeperMetadataServiceUri causes a NPE in WorkerUtils.initializeDlogNamespace which is a bug in Pulsar
+  # omit setting bookkeeperMetadataServiceUri until the bug is fixed when functions are enabled.
+  # bookkeeperMetadataServiceUri will default to configurationMetadataStoreUrl + "/ledgers" in that case
+  {{- if not .Values.components.functions }}
+  bookkeeperMetadataServiceUri: "metadata-store:{{ $configMetadataStoreUrl }}/ledgers"
   {{- end }}
   {{- end }}
   {{- if .Values.components.oxia }}
@@ -43,11 +50,35 @@ data:
   bookkeeperMetadataServiceUri: "{{ template "pulsar.oxia.metadata.url.bookkeeper" . }}"
   {{- end }}
 
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreAllowReadOnlyOperations" }}
+  PULSAR_PREFIX_metadataStoreAllowReadOnlyOperations: "{{ .Values.pulsar_metadata.metadataStoreAllowReadOnlyOperations }}"
+  {{- end }}
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreSessionTimeoutMillis" }}
+  metadataStoreSessionTimeoutMillis: "{{ .Values.pulsar_metadata.metadataStoreSessionTimeoutMillis }}"
+  {{- end }}
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreOperationTimeoutSeconds" }}
+  metadataStoreOperationTimeoutSeconds: "{{ .Values.pulsar_metadata.metadataStoreOperationTimeoutSeconds }}"
+  {{- end }}
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreCacheExpirySeconds" }}
+  metadataStoreCacheExpirySeconds: "{{ .Values.pulsar_metadata.metadataStoreCacheExpirySeconds }}"
+  {{- end }}
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreBatchingEnabled" }}
+  metadataStoreBatchingEnabled: "{{ .Values.pulsar_metadata.metadataStoreBatchingEnabled }}"
+  {{- end }}
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreBatchingMaxDelayMillis" }}
+  metadataStoreBatchingMaxDelayMillis: "{{ .Values.pulsar_metadata.metadataStoreBatchingMaxDelayMillis }}"
+  {{- end }}
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreBatchingMaxOperations" }}
+  metadataStoreBatchingMaxOperations: "{{ .Values.pulsar_metadata.metadataStoreBatchingMaxOperations }}"
+  {{- end }}
+  {{- if hasKey .Values.pulsar_metadata "metadataStoreBatchingMaxSizeKb" }}
+  metadataStoreBatchingMaxSizeKb: "{{ .Values.pulsar_metadata.metadataStoreBatchingMaxSizeKb }}"
+  {{- end }}
+
   # Broker settings
   clusterName: {{ template "pulsar.cluster.name" . }}
   exposeTopicLevelMetricsInPrometheus: "true"
   numHttpServerThreads: "8"
-  zooKeeperSessionTimeoutMillis: "30000"
   statusFilePath: "{{ template "pulsar.home" . }}/logs/status"
 
   # Tiered storage settings

charts/pulsar/values.yaml

@@ -880,6 +880,44 @@ pulsar_metadata:
   ## Timeout for running metadata initialization
   initTimeout: 60
 
+  ## Allow read-only operations on the metadata store when the metadata store is not available.
+  ## This is useful when you want to continue serving requests even if the metadata store is not fully available with quorum.
+  metadataStoreAllowReadOnlyOperations: false
+
+  ## The session timeout for the metadata store in milliseconds.
+  metadataStoreSessionTimeoutMillis: 30000
+
+  ## Metadata store operation timeout in seconds.
+  metadataStoreOperationTimeoutSeconds: 30
+
+  ## The expiry time for the metadata store cache in seconds.
+  metadataStoreCacheExpirySeconds: 300
+
+  ## Whether we should enable metadata operations batching
+  metadataStoreBatchingEnabled: true
+
+  ## Maximum delay to impose on batching grouping (in milliseconds)
+  metadataStoreBatchingMaxDelayMillis: 5
+
+  ## Maximum number of operations to include in a singular batch
+  metadataStoreBatchingMaxOperations: 1000
+
+  ## Maximum size of a batch (in KB)
+  metadataStoreBatchingMaxSizeKb: 128
+
+  ## BookKeeper metadata configuration settings with Pulsar Helm Chart deployments
+  bookkeeper:
+    ## Controls whether to use the PIP-45 metadata driver (PulsarMetadataBookieDriver) for BookKeeper components
+    ## when using ZooKeeper as a metadata store.
+    ## This is a global setting that applies to all BookKeeper components.
+    ## When set to true, BookKeeper components will use the PIP-45 metadata driver (PulsarMetadataBookieDriver).
+    ## When set to false, BookKeeper components will use BookKeeper's default ZooKeeper connection implementation.
+    usePulsarMetadataBookieDriver: false
+
+    ## The session timeout for the metadata store in milliseconds. This setting is mapped to `zkTimeout` in `bookkeeper.conf`.
+    ## due to implementation details in the PulsarMetadataBookieDriver, it also applies when Oxia metadata store is enabled.
+    metadataStoreSessionTimeoutMillis: 30000
+
   # resources for bin/pulsar initialize-cluster-metadata
   resources:
 #    requests: