Support schema-aware context publish in Go SDK

.ci/clusters/kafka-schema-registry.yaml

@@ -0,0 +1,48 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: my-schema-registry
+  labels:
+    app: my-schema-registry
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: my-schema-registry
+  template:
+    metadata:
+      labels:
+        app: my-schema-registry
+    spec:
+      containers:
+        - name: my-schema-registry
+          image: confluentinc/cp-schema-registry:7.8.0
+          ports:
+            - containerPort: 8081
+          env:
+            - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
+              value: PLAINTEXT://my-kafka.default.svc.cluster.local:9092
+            - name: SCHEMA_REGISTRY_LISTENERS
+              value: http://0.0.0.0:8081
+            - name: SCHEMA_REGISTRY_HOST_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+            - name: SCHEMA_REGISTRY_SCHEMA_REGISTRY_GROUP_ID
+              value: my-schema-registry-group
+            - name: SCHEMA_REGISTRY_KAFKASTORE_TOPIC
+              value: _schemas
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: my-schema-registry
+  labels:
+    app: my-schema-registry
+spec:
+  type: ClusterIP
+  ports:
+    - port: 8081
+      targetPort: 8081
+  selector:
+    app: my-schema-registry

.ci/clusters/kafka-values.yaml

@@ -0,0 +1,44 @@
+image:
+  repository: bitnamilegacy/kafka
+
+listeners:
+  client:
+    protocol: PLAINTEXT
+  controller:
+    protocol: PLAINTEXT
+  interbroker:
+    protocol: PLAINTEXT
+  external:
+    protocol: PLAINTEXT
+
+controller:
+  replicaCount: 1
+  controllerOnly: false
+  resourcesPreset: small
+
+  persistence:
+    enabled: false
+  logPersistence:
+    enabled: false
+
+broker:
+  replicaCount: 0
+
+service:
+  type: ClusterIP
+
+externalAccess:
+  enabled: false
+
+networkPolicy:
+  enabled: false
+
+overrideConfiguration:
+  auto.create.topics.enable: "true"
+
+  offsets.topic.replication.factor: 1
+  transaction.state.log.replication.factor: 1
+  transaction.state.log.min.isr: 1
+  default.replication.factor: 1
+  min.insync.replicas: 1
+  num.partitions: 3

.ci/examples/consume_avro.py

@@ -0,0 +1,46 @@
+import pulsar
+from pulsar.schema import *
+
+from _pulsar import InitialPosition
+
+import sys
+
+topic = sys.argv[1]
+expected_name = sys.argv[2] if len(sys.argv) > 2 else None
+expected_age = int(sys.argv[3]) if len(sys.argv) > 3 else None
+expected_grade = int(sys.argv[4]) if len(sys.argv) > 4 else None
+
+
+class Student(Record):
+    def __init__(self, name, age, grade):
+        self.name = name
+        self.age = age
+        self.grade = grade
+
+    name = String()
+    age = Integer()
+    grade = Integer()
+
+
+client = pulsar.Client('pulsar://localhost:6650')
+
+schema = pulsar.schema.AvroSchema(Student)
+consumer = client.subscribe(topic=topic,
+                            subscription_name='my-avro-subscription',
+                            crypto_key_reader=None,
+                            initial_position=InitialPosition.Earliest,
+                            schema=schema)
+
+try:
+    msg = consumer.receive(1000)
+    value = msg.value()
+    if expected_name is not None:
+        if value.name != expected_name or value.age != expected_age or value.grade != expected_grade:
+            print(value)
+            sys.exit(1)
+        print("schema message matches")
+    else:
+        print(value)
+except Exception:
+    if expected_name is not None:
+        sys.exit(1)

.ci/examples/contextPublishAvroGo/context_publish_avro.go

@@ -0,0 +1,143 @@
+//
+// 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.
+//
+
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/linkedin/goavro/v2"
+	"github.com/streamnative/pulsar-function-go/pf"
+)
+
+const studentAvroSchema = `{"type":"record","name":"Student","fields":[{"name":"name","type":["null","string"]},{"name":"age","type":["null","int"]},{"name":"grade","type":["null","int"]}]}`
+const studentAvroInputFallbackSchema = `{"type":"record","name":"Student","fields":[{"name":"name","type":["null","string"]},{"name":"age","type":"int"},{"name":"grade","type":"int"}]}`
+
+var studentCodec = mustAvroCodec(studentAvroSchema)
+var studentInputFallbackCodec = mustAvroCodec(studentAvroInputFallbackSchema)
+
+func mustAvroCodec(schema string) *goavro.Codec {
+	codec, err := goavro.NewCodec(schema)
+	if err != nil {
+		panic(err)
+	}
+	return codec
+}
+
+func HandleContextPublishAvro(ctx context.Context, in []byte) error {
+	record, err := decodeStudent(in)
+	if err != nil {
+		return err
+	}
+
+	name, err := avroUnionString(record["name"])
+	if err != nil {
+		return err
+	}
+	age, err := avroUnionInt(record["age"])
+	if err != nil {
+		return err
+	}
+	grade, err := avroUnionInt(record["grade"])
+	if err != nil {
+		return err
+	}
+
+	outputRecord := map[string]interface{}{
+		"name":  map[string]interface{}{"string": name},
+		"age":   map[string]interface{}{"int": age},
+		"grade": map[string]interface{}{"int": grade + 1},
+	}
+	output, err := studentCodec.BinaryFromNative(nil, outputRecord)
+	if err != nil {
+		return err
+	}
+
+	fc, ok := pf.FromContext(ctx)
+	if !ok {
+		return fmt.Errorf("missing function context")
+	}
+
+	publishTopic, ok := fc.GetUserConfValue("publishTopic").(string)
+	if !ok || publishTopic == "" {
+		return fmt.Errorf("missing publishTopic user config")
+	}
+
+	_, err = fc.PublishWithSchema(publishTopic, output, &pf.PublishSchema{
+		SchemaType: pf.SchemaTypeAvro,
+		Name:       "Student",
+		SchemaData: []byte(studentAvroSchema),
+	})
+	return err
+}
+
+func decodeStudent(data []byte) (map[string]interface{}, error) {
+	for _, codec := range []*goavro.Codec{studentCodec, studentInputFallbackCodec} {
+		native, _, err := codec.NativeFromBinary(data)
+		if err != nil {
+			continue
+		}
+		record, ok := native.(map[string]interface{})
+		if !ok {
+			return nil, fmt.Errorf("expected Avro record, got %T", native)
+		}
+		return record, nil
+	}
+	_, _, err := studentCodec.NativeFromBinary(data)
+	return nil, err
+}
+
+func avroUnionString(value interface{}) (string, error) {
+	switch typed := value.(type) {
+	case map[string]interface{}:
+		raw, ok := typed["string"]
+		if !ok {
+			return "", fmt.Errorf("expected string union branch, got %v", typed)
+		}
+		return avroUnionString(raw)
+	case string:
+		return typed, nil
+	default:
+		return "", fmt.Errorf("expected Avro string, got %T", value)
+	}
+}
+
+func avroUnionInt(value interface{}) (int32, error) {
+	switch typed := value.(type) {
+	case map[string]interface{}:
+		raw, ok := typed["int"]
+		if !ok {
+			return 0, fmt.Errorf("expected int union branch, got %v", typed)
+		}
+		return avroUnionInt(raw)
+	case int32:
+		return typed, nil
+	case int:
+		return int32(typed), nil
+	case int64:
+		return int32(typed), nil
+	default:
+		return 0, fmt.Errorf("expected Avro int, got %T", value)
+	}
+}
+
+func main() {
+	pf.Start(HandleContextPublishAvro)
+}

.ci/examples/produce_avro.py

@@ -0,0 +1,32 @@
+import pulsar
+from pulsar.schema import *
+
+import sys
+
+topic = sys.argv[1]
+name = sys.argv[2]
+age = int(sys.argv[3])
+grade = int(sys.argv[4])
+
+client = pulsar.Client('pulsar://localhost:6650')
+
+
+class Student(Record):
+    def __init__(self, name, age, grade):
+        self.name = name
+        self.age = age
+        self.grade = grade
+
+    name = String()
+    age = Integer()
+    grade = Integer()
+
+
+schema = pulsar.schema.AvroSchema(Student)
+producer = client.create_producer(topic=topic,
+                                  schema=schema)
+
+student = Student(name, age, grade)
+producer.send(student)
+
+producer.close()