feat: support ConnectorCatalog for builtinConnector (#1211)

<!--
### Contribution Checklist

- Name the pull request in the form "[charts/<chart-name>] Title of the
pull request".
Skip *[charts/<chart-name>]* if the PR doesn't change a specific chart.
E.g. `[docs] Fix typo in README`.

- Fill out the template below to describe the changes contributed by the
pull request. That will give reviewers the context they need to do the
review.

- Each pull request should address only one issue, not mix up code from
multiple issues.

  - Each commit in the pull request has a meaningful commit message

- Once all items of the checklist are addressed, remove the above text
and this checklist, leaving only the filled out template below.

**(The sections below can be removed for hotfixes of typos)**
-->

*(If this PR fixes a github issue, please add `Fixes #<xyz>`.)*

Fixes #<xyz>

*(or if this PR is one task of a github issue, please add `Master Issue:
#<xyz>` to link to the master issue.)*

Master Issue: #<xyz>

### Motivation

*Explain here the context, and why you're making that change. What is
the problem you're trying to solve.*

### Modifications

*Describe the modifications you've done.*

### Verifying this change

- [x] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test
coverage.

*(or)*

This change is already covered by existing tests, such as *(please
describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
- *Added integration tests for end-to-end deployment with large payloads
(10MB)*
  - *Extended integration test for recovery after broker failure*

### Documentation

Check the box below.

Need to update docs?

- [ ] `doc-required`

  (If you need help on updating docs, create a doc issue)

- [x] `no-need-doc`

  (Please explain why)

- [ ] `doc`

  (If this PR contains doc changes)

---------

Signed-off-by: ericsyh <[email protected]>
Co-authored-by: Rui Fu <[email protected]>
(cherry picked from commit 102aa0a)

Author: ericsyh

charts/sn-platform-slim/templates/broker/_functions.tpl

@@ -18,4 +18,28 @@ The pulsar root directory of functions image
 {{- else }}
 {{- template "pulsar.home" . }}
 {{- end }}
-{{- end }}
+{{- end }}
+
+{{/*
+The customWorkerConfig tpl
+*/}}
+{{- define "pulsar.functions.config.customWorkerConfig" -}}
+functionsWorkerServiceCustomConfigs:
+{{- if and .Values.broker.functionmesh.enabled (eq .Values.broker.functionmesh.builtinConnectorType "ConnectorCatalog") }}
+  connectorCatalogNamespace: "{{ template "pulsar.namespace" . }}"
+{{- end }}
+{{- end }}
+
+{{/*
+Merge customWorkerConfig tpl
+*/}}
+{{- define "pulsar.functions.config.mergedCustomWorkerConfig" -}}
+{{- $baseCustomWorkerConfig := include "pulsar.functions.config.customWorkerConfig" . | fromYaml -}}
+{{- if .Values.broker.functionmesh.customWorkerConfig }}
+{{- $dictCustomWorkerConfig := .Values.broker.functionmesh.customWorkerConfig | fromYaml -}}
+{{- $mergedConfig := mustMergeOverwrite (dict) $dictCustomWorkerConfig $baseCustomWorkerConfig -}}
+{{ toYaml $dictCustomWorkerConfig | quote }}
+{{- else -}}
+{{ $baseCustomWorkerConfig | toYaml | quote }}
+{{- end }}
+{{- end }}

charts/sn-platform-slim/templates/broker/broker-builtin-connectors.yaml

@@ -212,15 +212,21 @@ spec:
     function:
       enabled: true
       mesh:
+        {{- if eq .Values.broker.functionmesh.builtinConnectorType "ConfigMap" }}
         builtinConnectorsRef:
           name: {{ include "pulsar.builtinConnectorsConfigMapName" . }}
+        {{- end }}
         functionEnabled: {{ .Values.broker.functionmesh.mesh.functionEnabled }}
         sinkEnabled: {{ .Values.broker.functionmesh.mesh.sinkEnabled }}
         sourceEnabled: {{ .Values.broker.functionmesh.mesh.sourceEnabled }}
         uploadEnabled: {{ .Values.broker.functionmesh.mesh.uploadEnabled }}
+      {{- if and .Values.broker.functionmesh.enabled (eq .Values.broker.functionmesh.builtinConnectorType "ConnectorCatalog") }}
+      customWorkerConfig: {{ template "pulsar.functions.config.mergedCustomWorkerConfig" . }}
+      {{- else }}
       {{- if .Values.broker.functionmesh.customWorkerConfig }}
       customWorkerConfig: {{ .Values.broker.functionmesh.customWorkerConfig | quote }}
       {{- end }}
+      {{- end }}
       {{- if .Values.broker.functionmesh.custom }}
       custom:
 {{ toYaml .Values.broker.functionmesh.custom | indent 8 }}

charts/sn-platform-slim/templates/broker/broker-cluster.yaml

@@ -16,7 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-{{- if and .Values.components.broker .Values.broker.functionmesh.enabled}}
+{{- if and .Values.components.broker .Values.broker.functionmesh.enabled (eq .Values.broker.functionmesh.builtinConnectorType "ConfigMap")}}
 ---
 apiVersion: v1
 kind: ConfigMap

charts/sn-platform-slim/templates/broker/broker-configmap-functionmesh.yaml

@@ -1156,6 +1156,10 @@ broker:
       sourceEnabled: true
       sinkEnabled: true
     builtinConnectorRepository:
+    # The resource type for builtinConnector. Supported values: ConfigMap and ConnectorCatalog
+    # For ConnectorCatalog builtinConnectorType, make sure you are running with the sn-operator: https://github.com/streamnative/private-cloud
+    # And has ConnectorCatalog CRD
+    builtinConnectorType: ConfigMap
     customWorkerConfig:
     runnerImages:
     labels:

charts/sn-platform-slim/values.yaml

@@ -18,4 +18,28 @@ The pulsar root directory of functions image
 {{- else }}
 {{- template "pulsar.home" . }}
 {{- end }}
-{{- end }}
+{{- end }}
+
+{{/*
+The customWorkerConfig tpl
+*/}}
+{{- define "pulsar.functions.config.customWorkerConfig" -}}
+functionsWorkerServiceCustomConfigs:
+{{- if and .Values.broker.functionmesh.enabled (eq .Values.broker.functionmesh.builtinConnectorType "ConnectorCatalog") }}
+  connectorCatalogNamespace: "{{ template "pulsar.namespace" . }}"
+{{- end }}
+{{- end }}
+
+{{/*
+Merge customWorkerConfig tpl
+*/}}
+{{- define "pulsar.functions.config.mergedCustomWorkerConfig" -}}
+{{- $baseCustomWorkerConfig := include "pulsar.functions.config.customWorkerConfig" . | fromYaml -}}
+{{- if .Values.broker.functionmesh.customWorkerConfig }}
+{{- $dictCustomWorkerConfig := .Values.broker.functionmesh.customWorkerConfig | fromYaml -}}
+{{- $mergedConfig := mustMergeOverwrite (dict) $dictCustomWorkerConfig $baseCustomWorkerConfig -}}
+{{ toYaml $dictCustomWorkerConfig | quote }}
+{{- else -}}
+{{ $baseCustomWorkerConfig | toYaml | quote }}
+{{- end }}
+{{- end }}