Merge pull request #6 from streamnative/update

feat: update logic to adopt changes in pulsar-functions-generic-runtime

Author: jiangpengcheng

.ci/clusters/oxia.yaml

@@ -0,0 +1,37 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: oxia
+  labels:
+    name: oxia
+spec:
+  ports:
+    - port: 6648
+  clusterIP: None
+  selector:
+    role: oxia
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: oxia
+spec:
+  selector:
+    matchLabels:
+      role: oxia
+  serviceName: "oxia"
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        role: oxia
+    spec:
+      terminationGracePeriodSeconds: 10
+      containers:
+        - name: oxia
+          image: streamnative/oxia:0.13
+          command:
+            - oxia
+            - standalone
+          ports:
+            - containerPort: 6648

.ci/clusters/secret.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: test-secret
+type: Opaque
+data:
+  key: c2VjLXZhbHVl

.ci/clusters/values_skywalking_e2e_cluster.yaml

@@ -0,0 +1,129 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+## deployed withh emptyDir
+volumes:
+  persistence: false
+
+# disabled AntiAffinity
+affinity:
+  anti_affinity: false
+
+# disable auto recovery
+components:
+  autorecovery: false
+  pulsar_manager: false
+  sql_worker: false
+  proxy: false
+
+## disable monitoring stack
+monitoring:
+  # monitoring - prometheus
+  prometheus: false
+  # monitoring - grafana
+  grafana: false
+  # monitoring - node_exporter
+  node_exporter: false
+
+images:
+  zookeeper:
+    repository: streamnative/sn-platform
+    tag: 2.9.2.23
+  bookie:
+    repository: streamnative/sn-platform
+    tag: 2.9.2.23
+  broker:
+    repository: streamnative/sn-platform
+    tag: 2.9.2.23
+  functions:
+    repository: streamnative/sn-platform
+    tag: 2.9.2.23
+
+zookeeper:
+  replicaCount: 1
+  resources:
+    requests:
+      memory: 256Mi
+      cpu: 50m
+
+bookkeeper:
+  replicaCount: 0
+  metadata:
+    image:
+      repository: streamnative/sn-platform
+      tag: 2.9.2.23
+  resources:
+    requests:
+      memory: 256Mi
+      cpu: 50m
+  configData:
+    PULSAR_PREFIX_autoRecoveryDaemonEnabled: "false"
+    PULSAR_PREFIX_dlog.bkcEnsembleSize: "1"
+    PULSAR_PREFIX_dlog.bkcWriteQuorumSize: "1"
+    PULSAR_PREFIX_dlog.bkcAckQuorumSize: "1"
+    # `BOOKIE_MEM` is used for `bookie shell`
+    BOOKIE_MEM: >
+      -Xms128m
+      -Xmx256m
+      -XX:MaxDirectMemorySize=256m
+    # we use `bin/pulsar` for starting bookie daemons
+    PULSAR_MEM: >
+      -Xms128m
+      -Xmx256m
+      -XX:MaxDirectMemorySize=256m
+    PULSAR_GC: >
+      -XX:+UseG1GC
+      -XX:MaxGCPauseMillis=10
+      -XX:+ParallelRefProcEnabled
+      -XX:+UnlockExperimentalVMOptions
+      -XX:+AggressiveOpts
+      -XX:+DoEscapeAnalysis
+      -XX:ParallelGCThreads=4
+      -XX:ConcGCThreads=4
+      -XX:G1NewSizePercent=50
+      -XX:+DisableExplicitGC
+      -XX:-ResizePLAB
+      -XX:+ExitOnOutOfMemoryError
+      -XX:+PerfDisableSharedMem
+
+pulsar_metadata:
+  image:
+    repository: streamnative/sn-platform
+    tag: 2.9.2.23
+
+broker:
+  replicaCount: 1
+  configData:
+    ## Enable `autoSkipNonRecoverableData` since bookkeeper is running
+    ## without persistence
+    autoSkipNonRecoverableData: "true"
+    # storage settings
+    managedLedgerDefaultEnsembleSize: "1"
+    managedLedgerDefaultWriteQuorum: "1"
+    managedLedgerDefaultAckQuorum: "1"
+    enablePackagesManagement: "true"
+    PULSAR_PREFIX_enablePackagesManagement: "true"
+  resources:
+    requests:
+      memory: 256Mi
+      cpu: 50m
+
+functions:
+  functionState: true
+  useDedicatedRunner: false

.ci/examples/exclamationGo/exclamation.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/sirupsen/logrus"
+	"github.com/streamnative/pulsar-function-go/pf"
+)
+
+func HandleExclamation(ctx context.Context, in []byte) ([]byte, error) {
+	// 1. unmarshal []byte to your struct, use any schema you want
+	payload := string(in)
+
+	// 2. do your logic
+	if fc, ok := pf.FromContext(ctx); ok {
+		for _, word := range strings.Split(payload, " ") {
+			_ = fc.IncrCounter(word, 1)
+			count, _ := fc.GetCounter(word)
+			logrus.Infof("got word: %s for %d times", word, count)
+		}
+		// test user config and secrets
+		cfg := fc.GetUserConfValue("configKey")
+		sec, err := fc.GetSecret("secretKey")
+		if err == nil {
+			msg := fmt.Sprintf("config: %v, secret: %s", cfg, *sec)
+			_, _ = fc.Publish("persistent://public/default/test-exec-package-serde-extra", []byte(msg))
+		}
+	}
+	data := payload + "!"
+
+	// 3. marshal your struct to []byte
+	return []byte(data), nil
+
+}
+
+func main() {
+	pf.Start(HandleExclamation)
+}