diff --git a/.ci/clusters/kafka-schema-registry.yaml b/.ci/clusters/kafka-schema-registry.yaml
new file mode 100644
index 0000000..b27494a
--- /dev/null
+++ b/.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
diff --git a/.ci/clusters/kafka-values.yaml b/.ci/clusters/kafka-values.yaml
new file mode 100644
index 0000000..416170b
--- /dev/null
+++ b/.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
diff --git a/.ci/examples/consume_avro.py b/.ci/examples/consume_avro.py
new file mode 100644
index 0000000..27460f3
--- /dev/null
+++ b/.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)
diff --git a/.ci/examples/contextPublishAvroGo/context_publish_avro.go b/.ci/examples/contextPublishAvroGo/context_publish_avro.go
new file mode 100644
index 0000000..c9524f2
--- /dev/null
+++ b/.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.PublishMessageSchema{
+		SchemaType: pf.SchemaTypeAvro,
+		Name:       "Student",
+		SchemaData: 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)
+}
diff --git a/.ci/examples/produce_avro.py b/.ci/examples/produce_avro.py
new file mode 100644
index 0000000..43981d5
--- /dev/null
+++ b/.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()
diff --git a/.ci/helm.sh b/.ci/helm.sh
index 34b6b62..add0423 100644
--- a/.ci/helm.sh
+++ b/.ci/helm.sh
@@ -43,6 +43,18 @@ function ci::ensure_topic_with_schema() {
     fi
 }
 
+function ci::verify_topic_schema() {
+    topic=$1
+    schema=$2
+    expected_schema=$(echo "${schema}" | tr "[:lower:]" "[:upper:]")
+    schema_info=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin schemas get "${topic}") || return 1
+    echo "$schema_info"
+    if [[ "$schema_info" == *"\"type\" : \"${expected_schema}\""* ]] || [[ "$schema_info" == *"\"type\": \"${expected_schema}\""* ]] || [[ "$schema_info" == *"\"type\":\"${expected_schema}\""* ]]; then
+        return 0
+    fi
+    return 1
+}
+
 function ci::verify_function_mesh() {
     FUNCTION_NAME=$1
 
@@ -64,6 +76,96 @@ function ci::verify_function_mesh() {
     done
 }
 
+function ci::verify_kafka_function_mesh() {
+    FUNCTION_NAME=$1
+
+    num=0
+    while [[ ${num} -lt 1 ]]; do
+        sleep 5
+        kubectl get pods
+        num=$(kubectl get pods -l compute.functionmesh.io/name="${FUNCTION_NAME}" --no-headers 2>/dev/null | wc -l)
+    done
+
+    if ! kubectl wait -l compute.functionmesh.io/name="${FUNCTION_NAME}" --for=condition=Ready pod --timeout=2m; then
+        kubectl logs -l compute.functionmesh.io/name="${FUNCTION_NAME}" --tail=100 || true
+        return 1
+    fi
+    kubectl logs -l compute.functionmesh.io/name="${FUNCTION_NAME}" --tail=100 || true
+}
+
+function ci::kafka_client() {
+    command=$1
+    image=${KAFKA_CLIENT_IMAGE:-"confluentinc/cp-schema-registry:7.6.1"}
+    name="kafka-client-$(date +%s)-${RANDOM}"
+    kubectl run -n ${NAMESPACE} "${name}" --rm -i --restart=Never --image="${image}" --command -- bash -lc "${command}"
+}
+
+function ci::kafka_reset_topics() {
+    local topic
+    local deleted
+    local created
+    local bootstrap_server=${KAFKA_BOOTSTRAP_SERVER:-"my-kafka:9092"}
+    local kafka_admin_pod=${KAFKA_ADMIN_POD:-"my-kafka-controller-0"}
+    local kafka_topics_bin=${KAFKA_TOPICS_BIN:-"/opt/bitnami/kafka/bin/kafka-topics.sh"}
+    local partitions=${KAFKA_TOPIC_PARTITIONS:-"3"}
+    local replication_factor=${KAFKA_TOPIC_REPLICATION_FACTOR:-"1"}
+
+    for topic in "$@"; do
+        kubectl exec -n ${NAMESPACE} "${kafka_admin_pod}" -- "${kafka_topics_bin}" \
+            --bootstrap-server "${bootstrap_server}" --delete --if-exists --topic "${topic}" >/dev/null 2>&1 || true
+
+        deleted=0
+        for i in $(seq 1 12); do
+            if ! kubectl exec -n ${NAMESPACE} "${kafka_admin_pod}" -- "${kafka_topics_bin}" \
+                --bootstrap-server "${bootstrap_server}" --describe --topic "${topic}" >/dev/null 2>&1; then
+                deleted=1
+                break
+            fi
+            sleep 5
+        done
+        if [[ ${deleted} -ne 1 ]]; then
+            return 1
+        fi
+
+        created=0
+        for i in $(seq 1 12); do
+            if kubectl exec -n ${NAMESPACE} "${kafka_admin_pod}" -- "${kafka_topics_bin}" \
+                --bootstrap-server "${bootstrap_server}" --create --if-not-exists --topic "${topic}" \
+                --partitions "${partitions}" --replication-factor "${replication_factor}"; then
+                created=1
+                break
+            fi
+            sleep 5
+        done
+        if [[ ${created} -ne 1 ]]; then
+            return 1
+        fi
+    done
+}
+
+function ci::apply_manifest_or_debug() {
+    local manifests_file=$1
+    local apply_result
+
+    if apply_result=$(kubectl apply -f "${manifests_file}" 2>&1); then
+        echo "${apply_result}"
+        return 0
+    fi
+
+    echo "${apply_result}"
+    kubectl get pods -A || true
+    kubectl get events -A --sort-by='.lastTimestamp' | tail -100 || true
+    kubectl logs -n function-mesh -l app.kubernetes.io/name=function-mesh-operator --tail=200 || true
+    return 1
+}
+
+function ci::schema_registry_get_subject() {
+    subject=$1
+    image=${SCHEMA_REGISTRY_CURL_IMAGE:-"curlimages/curl:8.8.0"}
+    name="schema-registry-client-$(date +%s)-${RANDOM}"
+    kubectl run -n ${NAMESPACE} "${name}" --rm -i --restart=Never --image="${image}" --command -- sh -c "curl -sf http://my-schema-registry:8081/subjects/${subject}/versions/latest"
+}
+
 function ci::produce_message() {
     topic=$1
     message=$2
@@ -132,9 +234,9 @@ function ci::verify_avro_function() {
     timesleep=$7
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- python3 examples/produce_avro.py "${inputtopic}" "${name}" "${age}" "${grade}"
     sleep "$timesleep"
-    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- python3 examples/consume_avro.py "${outputtopic}")
+    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- python3 examples/consume_avro.py "${outputtopic}" "${name}" "${age}" "$((grade + 1))") || return 1
     echo "$MESSAGE"
-    if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
+    if [[ "$MESSAGE" == *"$outputmessage"* ]] || [[ "$MESSAGE" == *"schema message matches"* ]]; then
         return 0
     fi
     return 1
@@ -166,12 +268,20 @@ function ci::verify_topic_with_auth() {
 function ci::verify_total_processed() {
   path="$1-function-headless:9094/metrics"
   total=$2
-  curlCommand="kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'curl \"${path}\"' | grep pulsar_function_processed_successfully_total | grep -v \"#\""
-  MESSAGE=$(sh -c "$curlCommand")
+  curlCommand="kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'curl \"${path}\"' | grep 'pulsar_function_processed_successfully_total{' | grep -v \"#\""
+  MESSAGE=""
+  for i in $(seq 1 12); do
+      if MESSAGE=$(sh -c "$curlCommand" 2>&1); then
+          if [[ "$MESSAGE" == *" ${total}" ]]; then
+              echo "$MESSAGE"
+              return 0
+          fi
+      fi
+      if [[ ${i} -lt 12 ]]; then
+          sleep 5
+      fi
+  done
   echo "$MESSAGE"
-  if [[ "$MESSAGE" == *"$total" ]]; then
-      return 0
-  fi
   return 1
 }
 
@@ -180,3 +290,11 @@ function ci::create_topic() {
   partition=${2:-"1"}
   kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics create-partitioned-topic ${topic} -p ${partition} || true
 }
+
+function ci::pulsar_reset_topics() {
+  local topic
+  for topic in "$@"; do
+    kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics delete --force "${topic}" >/dev/null 2>&1 || true
+    kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin schemas delete "${topic}" >/dev/null 2>&1 || true
+  done
+}
diff --git a/.ci/tests/integration/cases/kafka-context-publish-avro/manifests.yaml b/.ci/tests/integration/cases/kafka-context-publish-avro/manifests.yaml
new file mode 100644
index 0000000..9f0d017
--- /dev/null
+++ b/.ci/tests/integration/cases/kafka-context-publish-avro/manifests.yaml
@@ -0,0 +1,53 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: kafka-context-publish-avro-generic-sample
+  namespace: default
+spec:
+  image: streamnative/pulsar-functions-generic-base-runner:latest
+  className: contextpublishavro
+  forwardSourceMessageProperty: true
+  maxPendingAsyncRequests: 1000
+  replicas: 1
+  input:
+    sourceSpecs:
+      "kafka-context-publish-avro-input":
+        schemaType: avro
+    typeClassName: Student
+  output:
+    topic: kafka-context-publish-avro-unused-output
+    typeClassName: string
+  funcConfig:
+    publishTopic: kafka-context-publish-avro-output
+    _child_protocol: binary_v2
+  resources:
+    requests:
+      cpu: 500m
+      memory: 0.5G
+    limits:
+      cpu: 500m
+      memory: 0.5G
+  pulsarPackageService:
+    pulsarConfig: "test-kafka-context-publish-avro-pulsar"
+    tlsConfig:
+      enabled: false
+      allowInsecure: true
+      hostnameVerification: true
+  kafka:
+    bootstrapServers: my-kafka.default.svc.cluster.local:9092
+    producerConfig:
+      schema.registry.url: http://my-schema-registry.default.svc.cluster.local:8081
+  genericRuntime:
+    functionFile: contextpublishavro
+    functionFileLocation: function://public/default/exec-context-publish-avro
+    language: executable
+  clusterName: test
+  autoAck: true
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-kafka-context-publish-avro-pulsar
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650
diff --git a/.ci/tests/integration/cases/kafka-context-publish-avro/verify.sh b/.ci/tests/integration/cases/kafka-context-publish-avro/verify.sh
new file mode 100644
index 0000000..e10bf3d
--- /dev/null
+++ b/.ci/tests/integration/cases/kafka-context-publish-avro/verify.sh
@@ -0,0 +1,93 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+set -e
+
+E2E_DIR=$(dirname "$0")
+BASE_DIR=$(cd "${E2E_DIR}"/../../../../..;pwd)
+E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"}
+
+source "${BASE_DIR}"/.ci/helm.sh
+
+if [ ! "$KUBECONFIG" ]; then
+  export KUBECONFIG=${E2E_KUBECONFIG}
+fi
+
+manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/kafka-context-publish-avro/manifests.yaml
+avro_schema='{"type":"record","name":"Student","fields":[{"name":"name","type":["null","string"]},{"name":"age","type":["null","int"]},{"name":"grade","type":["null","int"]}]}'
+
+if ! reset_topics_result=$(ci::kafka_reset_topics kafka-context-publish-avro-input kafka-context-publish-avro-output 2>&1); then
+  echo "$reset_topics_result"
+  exit 1
+fi
+
+if ! apply_result=$(ci::apply_manifest_or_debug "${manifests_file}" 2>&1); then
+  echo "$apply_result"
+  exit 1
+fi
+
+if ! verify_fm_result=$(ci::verify_kafka_function_mesh kafka-context-publish-avro-generic-sample 2>&1); then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+produce_command=$(cat <<EOF
+cat > /tmp/value.avsc <<'AVSC'
+${avro_schema}
+AVSC
+cat > /tmp/message.json <<'JSON'
+{"name":{"string":"jack"},"age":{"int":21},"grade":{"int":80}}
+JSON
+kafka-avro-console-producer --bootstrap-server my-kafka:9092 --topic kafka-context-publish-avro-input --property schema.registry.url=http://my-schema-registry:8081 --property value.schema.file=/tmp/value.avsc < /tmp/message.json
+EOF
+)
+if ! produce_result=$(ci::kafka_client "${produce_command}" 2>&1); then
+  echo "$produce_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+consume_command='timeout 60 kafka-avro-console-consumer --bootstrap-server my-kafka:9092 --topic kafka-context-publish-avro-output --from-beginning --max-messages 1 --property schema.registry.url=http://my-schema-registry:8081'
+if ! consume_result=$(ci::kafka_client "${consume_command}" 2>&1); then
+  echo "$consume_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+if [[ "$consume_result" != *'"name":{"string":"jack"}'* ]] || [[ "$consume_result" != *'"age":{"int":21}'* ]] || [[ "$consume_result" != *'"grade":{"int":81}'* ]]; then
+  echo "$consume_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+if ! schema_result=$(ci::schema_registry_get_subject "kafka-context-publish-avro-output-value" 2>&1); then
+  echo "$schema_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+if ! verify_total_processed=$(ci::verify_total_processed kafka-context-publish-avro-generic-sample 1); then
+  echo "$verify_total_processed"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+echo "e2e-test: ok" | yq eval -
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
diff --git a/.ci/tests/integration/cases/pulsar-context-publish-avro/manifests.yaml b/.ci/tests/integration/cases/pulsar-context-publish-avro/manifests.yaml
new file mode 100644
index 0000000..02ab1c3
--- /dev/null
+++ b/.ci/tests/integration/cases/pulsar-context-publish-avro/manifests.yaml
@@ -0,0 +1,49 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: pulsar-context-publish-avro-generic-sample
+  namespace: default
+spec:
+  image: streamnative/pulsar-functions-generic-base-runner:latest
+  className: contextpublishavro
+  forwardSourceMessageProperty: true
+  maxPendingAsyncRequests: 1000
+  replicas: 1
+  input:
+    sourceSpecs:
+      "persistent://public/default/pulsar-context-publish-avro-input":
+        schemaType: avro
+    typeClassName: Student
+  output:
+    topic: persistent://public/default/pulsar-context-publish-avro-unused-output
+    typeClassName: string
+  funcConfig:
+    publishTopic: persistent://public/default/pulsar-context-publish-avro-output
+    _child_protocol: binary_v2
+  resources:
+    requests:
+      cpu: 500m
+      memory: 0.5G
+    limits:
+      cpu: 500m
+      memory: 0.5G
+  pulsar:
+    pulsarConfig: "test-pulsar-context-publish-avro-pulsar"
+    tlsConfig:
+      enabled: false
+      allowInsecure: true
+      hostnameVerification: true
+  genericRuntime:
+    functionFile: contextpublishavro
+    functionFileLocation: function://public/default/exec-context-publish-avro
+    language: executable
+  clusterName: test
+  autoAck: true
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-pulsar-context-publish-avro-pulsar
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650
diff --git a/.ci/tests/integration/cases/pulsar-context-publish-avro/verify.sh b/.ci/tests/integration/cases/pulsar-context-publish-avro/verify.sh
new file mode 100644
index 0000000..e939b0b
--- /dev/null
+++ b/.ci/tests/integration/cases/pulsar-context-publish-avro/verify.sh
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+set -e
+
+E2E_DIR=$(dirname "$0")
+BASE_DIR=$(cd "${E2E_DIR}"/../../../../..;pwd)
+PULSAR_NAMESPACE=${PULSAR_NAMESPACE:-"default"}
+PULSAR_RELEASE_NAME=${PULSAR_RELEASE_NAME:-"sn-platform"}
+E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"}
+
+source "${BASE_DIR}"/.ci/helm.sh
+
+if [ ! "$KUBECONFIG" ]; then
+  export KUBECONFIG=${E2E_KUBECONFIG}
+fi
+
+manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/pulsar-context-publish-avro/manifests.yaml
+
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+ci::pulsar_reset_topics \
+  "persistent://public/default/pulsar-context-publish-avro-input" \
+  "persistent://public/default/pulsar-context-publish-avro-output" \
+  "persistent://public/default/pulsar-context-publish-avro-unused-output"
+
+kubectl apply -f "${manifests_file}" > /dev/null 2>&1
+
+if ! verify_fm_result=$(ci::verify_function_mesh pulsar-context-publish-avro-generic-sample 2>&1); then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+binary_safe_name=$'jack/slash\nline\rcarriage'
+if ! verify_go_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_avro_function "persistent://public/default/pulsar-context-publish-avro-input" "persistent://public/default/pulsar-context-publish-avro-output" "${binary_safe_name}" 21 80 "schema message matches" 10 2>&1); then
+  echo "$verify_go_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+if ! verify_output_schema=$(ci::verify_topic_schema "persistent://public/default/pulsar-context-publish-avro-output" "avro" 2>&1); then
+  echo "$verify_output_schema"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+if ! verify_total_processed=$(ci::verify_total_processed pulsar-context-publish-avro-generic-sample 1); then
+  echo "$verify_total_processed"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+echo "e2e-test: ok" | yq eval -
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
diff --git a/context.proto b/context.proto
index f7b3e54..e2ca361 100644
--- a/context.proto
+++ b/context.proto
@@ -74,6 +74,14 @@ message MessageId {
   string id = 1;
 }
 
+message PublishSchema {
+  string schemaType = 1;
+  string name = 2;
+  bytes schemaData = 3;
+  map<string, string> properties = 4;
+  string subject = 5;
+}
+
 message PulsarMessage {
   string topic = 1;
   bytes payload = 2;
@@ -86,6 +94,7 @@ message PulsarMessage {
   int32 eventTimestamp = 9;
   int32 deliverAt = 10;
   int32 deliverAfter = 11;
+  PublishSchema schema = 12;
 }
 
 message Record {
diff --git a/pf/context.go b/pf/context.go
index 3bf1f0b..28a195e 100644
--- a/pf/context.go
+++ b/pf/context.go
@@ -192,13 +192,56 @@ func (c *FunctionContext) GetSecret(secretName string) (*string, error) {
 
 // Publish publishes payload to the given topic
 func (c *FunctionContext) Publish(topic string, payload []byte) (*SendMessageId, error) {
+	return c.publish(topic, payload, nil)
+}
+
+// PublishMessageSchema describes schema metadata for messages sent through
+// PublishWithSchema.
+type PublishMessageSchema struct {
+	SchemaType string
+	Name       string
+	SchemaData string
+	Properties map[string]string
+	Subject    string
+}
+
+// PublishWithSchema publishes an already-encoded payload with schema metadata.
+//
+// The Go SDK does not encode the payload. The caller must marshal the message
+// according to schema before calling this method. This requires a generic
+// runtime runner that consumes PulsarMessage.schema, such as a runner build that
+// includes streamnative/pulsar-functions-generic-runtime#42. Older runners
+// silently ignore this proto3 field and publish the payload as raw bytes.
+func (c *FunctionContext) PublishWithSchema(topic string, payload []byte, schema PublishMessageSchema) (*SendMessageId, error) {
+	return c.publish(topic, payload, publishMessageSchemaToProto(schema))
+}
+
+func (c *FunctionContext) publish(topic string, payload []byte, schema *PublishSchema) (*SendMessageId, error) {
 	messageId, err := c.stub.Publish(c.ctx, &PulsarMessage{
 		Topic:   topic,
 		Payload: payload,
+		Schema:  schema,
 	})
 	return messageId, err
 }
 
+func publishMessageSchemaToProto(schema PublishMessageSchema) *PublishSchema {
+	if isNoSchemaType(schema.SchemaType) {
+		return nil
+	}
+	properties := schema.Properties
+	if properties == nil {
+		properties = map[string]string{}
+	}
+	return &PublishSchema{
+		SchemaType: schema.SchemaType,
+		Name:       schema.Name,
+		SchemaData: []byte(schema.SchemaData),
+		Properties: properties,
+		Subject:    schema.Subject,
+	}
+}
+
 // GetCurrentRecord gets the current message from the function context
 func (c *FunctionContext) GetCurrentRecord() (*Record, error) {
 	if c.message != nil {
diff --git a/pf/context.pb.go b/pf/context.pb.go
index 0e56e2d..c00c397 100644
--- a/pf/context.pb.go
+++ b/pf/context.pb.go
@@ -460,28 +460,108 @@ func (x *MessageId) GetId() string {
 	return ""
 }
 
+type PublishSchema struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	SchemaType string            `protobuf:"bytes,1,opt,name=schemaType,proto3" json:"schemaType,omitempty"`
+	Name       string            `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"`
+	SchemaData []byte            `protobuf:"bytes,3,opt,name=schemaData,proto3" json:"schemaData,omitempty"`
+	Properties map[string]string `protobuf:"bytes,4,rep,name=properties,proto3" json:"properties,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
+	Subject    string            `protobuf:"bytes,5,opt,name=subject,proto3" json:"subject,omitempty"`
+}
+
+func (x *PublishSchema) Reset() {
+	*x = PublishSchema{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_context_proto_msgTypes[8]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *PublishSchema) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PublishSchema) ProtoMessage() {}
+
+func (x *PublishSchema) ProtoReflect() protoreflect.Message {
+	mi := &file_context_proto_msgTypes[8]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use PublishSchema.ProtoReflect.Descriptor instead.
+func (*PublishSchema) Descriptor() ([]byte, []int) {
+	return file_context_proto_rawDescGZIP(), []int{8}
+}
+
+func (x *PublishSchema) GetSchemaType() string {
+	if x != nil {
+		return x.SchemaType
+	}
+	return ""
+}
+
+func (x *PublishSchema) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *PublishSchema) GetSchemaData() []byte {
+	if x != nil {
+		return x.SchemaData
+	}
+	return nil
+}
+
+func (x *PublishSchema) GetProperties() map[string]string {
+	if x != nil {
+		return x.Properties
+	}
+	return nil
+}
+
+func (x *PublishSchema) GetSubject() string {
+	if x != nil {
+		return x.Subject
+	}
+	return ""
+}
+
 type PulsarMessage struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Topic               string `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
-	Payload             []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"`
-	MessageId           string `protobuf:"bytes,3,opt,name=messageId,proto3" json:"messageId,omitempty"`
-	Properties          string `protobuf:"bytes,4,opt,name=properties,proto3" json:"properties,omitempty"`
-	PartitionKey        string `protobuf:"bytes,5,opt,name=partitionKey,proto3" json:"partitionKey,omitempty"`
-	SequenceId          string `protobuf:"bytes,6,opt,name=sequenceId,proto3" json:"sequenceId,omitempty"`
-	ReplicationClusters string `protobuf:"bytes,7,opt,name=replicationClusters,proto3" json:"replicationClusters,omitempty"`
-	DisableReplication  bool   `protobuf:"varint,8,opt,name=disableReplication,proto3" json:"disableReplication,omitempty"`
-	EventTimestamp      int32  `protobuf:"varint,9,opt,name=eventTimestamp,proto3" json:"eventTimestamp,omitempty"`
-	DeliverAt           int32  `protobuf:"varint,10,opt,name=deliverAt,proto3" json:"deliverAt,omitempty"`
-	DeliverAfter        int32  `protobuf:"varint,11,opt,name=deliverAfter,proto3" json:"deliverAfter,omitempty"`
+	Topic               string         `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
+	Payload             []byte         `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"`
+	MessageId           string         `protobuf:"bytes,3,opt,name=messageId,proto3" json:"messageId,omitempty"`
+	Properties          string         `protobuf:"bytes,4,opt,name=properties,proto3" json:"properties,omitempty"`
+	PartitionKey        string         `protobuf:"bytes,5,opt,name=partitionKey,proto3" json:"partitionKey,omitempty"`
+	SequenceId          string         `protobuf:"bytes,6,opt,name=sequenceId,proto3" json:"sequenceId,omitempty"`
+	ReplicationClusters string         `protobuf:"bytes,7,opt,name=replicationClusters,proto3" json:"replicationClusters,omitempty"`
+	DisableReplication  bool           `protobuf:"varint,8,opt,name=disableReplication,proto3" json:"disableReplication,omitempty"`
+	EventTimestamp      int32          `protobuf:"varint,9,opt,name=eventTimestamp,proto3" json:"eventTimestamp,omitempty"`
+	DeliverAt           int32          `protobuf:"varint,10,opt,name=deliverAt,proto3" json:"deliverAt,omitempty"`
+	DeliverAfter        int32          `protobuf:"varint,11,opt,name=deliverAfter,proto3" json:"deliverAfter,omitempty"`
+	Schema              *PublishSchema `protobuf:"bytes,12,opt,name=schema,proto3" json:"schema,omitempty"`
 }
 
 func (x *PulsarMessage) Reset() {
 	*x = PulsarMessage{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_context_proto_msgTypes[8]
+		mi := &file_context_proto_msgTypes[9]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -494,7 +574,7 @@ func (x *PulsarMessage) String() string {
 func (*PulsarMessage) ProtoMessage() {}
 
 func (x *PulsarMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_context_proto_msgTypes[8]
+	mi := &file_context_proto_msgTypes[9]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -507,7 +587,7 @@ func (x *PulsarMessage) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use PulsarMessage.ProtoReflect.Descriptor instead.
 func (*PulsarMessage) Descriptor() ([]byte, []int) {
-	return file_context_proto_rawDescGZIP(), []int{8}
+	return file_context_proto_rawDescGZIP(), []int{9}
 }
 
 func (x *PulsarMessage) GetTopic() string {
@@ -587,6 +667,13 @@ func (x *PulsarMessage) GetDeliverAfter() int32 {
 	return 0
 }
 
+func (x *PulsarMessage) GetSchema() *PublishSchema {
+	if x != nil {
+		return x.Schema
+	}
+	return nil
+}
+
 type Record struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -604,7 +691,7 @@ type Record struct {
 func (x *Record) Reset() {
 	*x = Record{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_context_proto_msgTypes[9]
+		mi := &file_context_proto_msgTypes[10]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -617,7 +704,7 @@ func (x *Record) String() string {
 func (*Record) ProtoMessage() {}
 
 func (x *Record) ProtoReflect() protoreflect.Message {
-	mi := &file_context_proto_msgTypes[9]
+	mi := &file_context_proto_msgTypes[10]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -630,7 +717,7 @@ func (x *Record) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Record.ProtoReflect.Descriptor instead.
 func (*Record) Descriptor() ([]byte, []int) {
-	return file_context_proto_rawDescGZIP(), []int{9}
+	return file_context_proto_rawDescGZIP(), []int{10}
 }
 
 func (x *Record) GetPayload() []byte {
@@ -694,7 +781,7 @@ type MetricData struct {
 func (x *MetricData) Reset() {
 	*x = MetricData{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_context_proto_msgTypes[10]
+		mi := &file_context_proto_msgTypes[11]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -707,7 +794,7 @@ func (x *MetricData) String() string {
 func (*MetricData) ProtoMessage() {}
 
 func (x *MetricData) ProtoReflect() protoreflect.Message {
-	mi := &file_context_proto_msgTypes[10]
+	mi := &file_context_proto_msgTypes[11]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -720,7 +807,7 @@ func (x *MetricData) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use MetricData.ProtoReflect.Descriptor instead.
 func (*MetricData) Descriptor() ([]byte, []int) {
-	return file_context_proto_rawDescGZIP(), []int{10}
+	return file_context_proto_rawDescGZIP(), []int{11}
 }
 
 func (x *MetricData) GetMetricName() string {
@@ -750,7 +837,7 @@ type Partition struct {
 func (x *Partition) Reset() {
 	*x = Partition{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_context_proto_msgTypes[11]
+		mi := &file_context_proto_msgTypes[12]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -763,7 +850,7 @@ func (x *Partition) String() string {
 func (*Partition) ProtoMessage() {}
 
 func (x *Partition) ProtoReflect() protoreflect.Message {
-	mi := &file_context_proto_msgTypes[11]
+	mi := &file_context_proto_msgTypes[12]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -776,7 +863,7 @@ func (x *Partition) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Partition.ProtoReflect.Descriptor instead.
 func (*Partition) Descriptor() ([]byte, []int) {
-	return file_context_proto_rawDescGZIP(), []int{11}
+	return file_context_proto_rawDescGZIP(), []int{12}
 }
 
 func (x *Partition) GetTopicName() string {
@@ -832,100 +919,119 @@ var file_context_proto_rawDesc = []byte{
 	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f,
 	0x61, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61,
 	0x64, 0x22, 0x1b, 0x0a, 0x09, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x12, 0x0e,
-	0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x8d,
-	0x03, 0x0a, 0x0d, 0x50, 0x75, 0x6c, 0x73, 0x61, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
-	0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61,
-	0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64,
-	0x12, 0x1c, 0x0a, 0x09, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x18, 0x03, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x09, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x12, 0x1e,
-	0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x18, 0x04, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x12, 0x22,
-	0x0a, 0x0c, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x4b, 0x65, 0x79, 0x18, 0x05,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x4b,
-	0x65, 0x79, 0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, 0x65, 0x49, 0x64,
-	0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, 0x65,
-	0x49, 0x64, 0x12, 0x30, 0x0a, 0x13, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
-	0x6e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x73, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x13, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6c, 0x75, 0x73,
-	0x74, 0x65, 0x72, 0x73, 0x12, 0x2e, 0x0a, 0x12, 0x64, 0x69, 0x73, 0x61, 0x62, 0x6c, 0x65, 0x52,
-	0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x08, 0x20, 0x01, 0x28, 0x08,
-	0x52, 0x12, 0x64, 0x69, 0x73, 0x61, 0x62, 0x6c, 0x65, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61,
-	0x74, 0x69, 0x6f, 0x6e, 0x12, 0x26, 0x0a, 0x0e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d,
-	0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x09, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0e, 0x65, 0x76,
-	0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x1c, 0x0a, 0x09,
-	0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x41, 0x74, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x05, 0x52,
-	0x09, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x41, 0x74, 0x12, 0x22, 0x0a, 0x0c, 0x64, 0x65,
-	0x6c, 0x69, 0x76, 0x65, 0x72, 0x41, 0x66, 0x74, 0x65, 0x72, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x05,
-	0x52, 0x0c, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x41, 0x66, 0x74, 0x65, 0x72, 0x22, 0xda,
-	0x01, 0x0a, 0x06, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79,
-	0x6c, 0x6f, 0x61, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c,
-	0x6f, 0x61, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49,
-	0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x18,
-	0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65,
-	0x73, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03,
-	0x6b, 0x65, 0x79, 0x12, 0x20, 0x0a, 0x0b, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e,
-	0x49, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74,
-	0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e, 0x61,
-	0x6d, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e,
-	0x61, 0x6d, 0x65, 0x12, 0x26, 0x0a, 0x0e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65,
-	0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0e, 0x65, 0x76, 0x65,
-	0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x22, 0x42, 0x0a, 0x0a, 0x4d,
-	0x65, 0x74, 0x72, 0x69, 0x63, 0x44, 0x61, 0x74, 0x61, 0x12, 0x1e, 0x0a, 0x0a, 0x6d, 0x65, 0x74,
-	0x72, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x6d,
-	0x65, 0x74, 0x72, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c,
-	0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x01, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22,
-	0x6f, 0x0a, 0x09, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1c, 0x0a, 0x09,
-	0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x09, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x26, 0x0a, 0x0e, 0x70, 0x61,
-	0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x18, 0x02, 0x20, 0x01,
-	0x28, 0x05, 0x52, 0x0e, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64,
-	0x65, 0x78, 0x12, 0x1c, 0x0a, 0x09, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x18,
-	0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64,
-	0x32, 0x90, 0x05, 0x0a, 0x0e, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x53, 0x65, 0x72, 0x76,
-	0x69, 0x63, 0x65, 0x12, 0x3b, 0x0a, 0x07, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x16,
-	0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x50, 0x75, 0x6c, 0x73, 0x61, 0x72, 0x4d,
-	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x16, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74,
-	0x2e, 0x53, 0x65, 0x6e, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x22, 0x00,
-	0x12, 0x36, 0x0a, 0x0d, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x63, 0x6f, 0x72,
-	0x64, 0x12, 0x12, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x4d, 0x65, 0x73, 0x73,
-	0x61, 0x67, 0x65, 0x49, 0x64, 0x1a, 0x0f, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e,
-	0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x22, 0x00, 0x12, 0x3e, 0x0a, 0x0d, 0x72, 0x65, 0x63, 0x6f,
-	0x72, 0x64, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x12, 0x13, 0x2e, 0x63, 0x6f, 0x6e, 0x74,
-	0x65, 0x78, 0x74, 0x2e, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x44, 0x61, 0x74, 0x61, 0x1a, 0x16,
-	0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66,
-	0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x34, 0x0a, 0x04, 0x73, 0x65, 0x65, 0x6b,
-	0x12, 0x12, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x50, 0x61, 0x72, 0x74, 0x69,
-	0x74, 0x69, 0x6f, 0x6e, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x35,
-	0x0a, 0x05, 0x70, 0x61, 0x75, 0x73, 0x65, 0x12, 0x12, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78,
+	0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x84,
+	0x02, 0x0a, 0x0d, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61,
+	0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x54, 0x79, 0x70, 0x65, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x54, 0x79, 0x70, 0x65,
+	0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
+	0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x44, 0x61,
+	0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0a, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61,
+	0x44, 0x61, 0x74, 0x61, 0x12, 0x46, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69,
+	0x65, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x26, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65,
+	0x78, 0x74, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61,
+	0x2e, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79,
+	0x52, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x12, 0x18, 0x0a, 0x07,
+	0x73, 0x75, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73,
+	0x75, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x1a, 0x3d, 0x0a, 0x0f, 0x50, 0x72, 0x6f, 0x70, 0x65, 0x72,
+	0x74, 0x69, 0x65, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76,
+	0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75,
+	0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xbd, 0x03, 0x0a, 0x0d, 0x50, 0x75, 0x6c, 0x73, 0x61, 0x72,
+	0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x18, 0x0a,
+	0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07,
+	0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6d, 0x65, 0x73, 0x73, 0x61,
+	0x67, 0x65, 0x49, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6d, 0x65, 0x73, 0x73,
+	0x61, 0x67, 0x65, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65, 0x72, 0x74,
+	0x69, 0x65, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x70, 0x65,
+	0x72, 0x74, 0x69, 0x65, 0x73, 0x12, 0x22, 0x0a, 0x0c, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69,
+	0x6f, 0x6e, 0x4b, 0x65, 0x79, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x70, 0x61, 0x72,
+	0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x4b, 0x65, 0x79, 0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x65, 0x71,
+	0x75, 0x65, 0x6e, 0x63, 0x65, 0x49, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73,
+	0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, 0x65, 0x49, 0x64, 0x12, 0x30, 0x0a, 0x13, 0x72, 0x65, 0x70,
+	0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x73,
+	0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74,
+	0x69, 0x6f, 0x6e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x73, 0x12, 0x2e, 0x0a, 0x12, 0x64,
+	0x69, 0x73, 0x61, 0x62, 0x6c, 0x65, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f,
+	0x6e, 0x18, 0x08, 0x20, 0x01, 0x28, 0x08, 0x52, 0x12, 0x64, 0x69, 0x73, 0x61, 0x62, 0x6c, 0x65,
+	0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x26, 0x0a, 0x0e, 0x65,
+	0x76, 0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x09, 0x20,
+	0x01, 0x28, 0x05, 0x52, 0x0e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74,
+	0x61, 0x6d, 0x70, 0x12, 0x1c, 0x0a, 0x09, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x41, 0x74,
+	0x18, 0x0a, 0x20, 0x01, 0x28, 0x05, 0x52, 0x09, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x41,
+	0x74, 0x12, 0x22, 0x0a, 0x0c, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x41, 0x66, 0x74, 0x65,
+	0x72, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0c, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72,
+	0x41, 0x66, 0x74, 0x65, 0x72, 0x12, 0x2e, 0x0a, 0x06, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x18,
+	0x0c, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e,
+	0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x52, 0x06, 0x73,
+	0x63, 0x68, 0x65, 0x6d, 0x61, 0x22, 0xda, 0x01, 0x0a, 0x06, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64,
+	0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x0c, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6d, 0x65,
+	0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6d,
+	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x70,
+	0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x70, 0x72,
+	0x6f, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18,
+	0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x20, 0x0a, 0x0b, 0x70, 0x61,
+	0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x0b, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x1c, 0x0a, 0x09,
+	0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x09, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x26, 0x0a, 0x0e, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x07, 0x20, 0x01,
+	0x28, 0x03, 0x52, 0x0e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61,
+	0x6d, 0x70, 0x22, 0x42, 0x0a, 0x0a, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x44, 0x61, 0x74, 0x61,
+	0x12, 0x1e, 0x0a, 0x0a, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65,
+	0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x01, 0x52,
+	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x6f, 0x0a, 0x09, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74,
+	0x69, 0x6f, 0x6e, 0x12, 0x1c, 0x0a, 0x09, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e, 0x61, 0x6d, 0x65,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x4e, 0x61, 0x6d,
+	0x65, 0x12, 0x26, 0x0a, 0x0e, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e,
+	0x64, 0x65, 0x78, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0e, 0x70, 0x61, 0x72, 0x74, 0x69,
+	0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x12, 0x1c, 0x0a, 0x09, 0x6d, 0x65, 0x73,
+	0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x6d, 0x65,
+	0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x32, 0x90, 0x05, 0x0a, 0x0e, 0x43, 0x6f, 0x6e, 0x74,
+	0x65, 0x78, 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x3b, 0x0a, 0x07, 0x70, 0x75,
+	0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x16, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e,
+	0x50, 0x75, 0x6c, 0x73, 0x61, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x16, 0x2e,
+	0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x4d, 0x65, 0x73, 0x73,
+	0x61, 0x67, 0x65, 0x49, 0x64, 0x22, 0x00, 0x12, 0x36, 0x0a, 0x0d, 0x63, 0x75, 0x72, 0x72, 0x65,
+	0x6e, 0x74, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x12, 0x12, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65,
+	0x78, 0x74, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x1a, 0x0f, 0x2e, 0x63,
+	0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x22, 0x00, 0x12,
+	0x3e, 0x0a, 0x0d, 0x72, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73,
+	0x12, 0x13, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x4d, 0x65, 0x74, 0x72, 0x69,
+	0x63, 0x44, 0x61, 0x74, 0x61, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12,
+	0x34, 0x0a, 0x04, 0x73, 0x65, 0x65, 0x6b, 0x12, 0x12, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78,
 	0x74, 0x2e, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x1a, 0x16, 0x2e, 0x67, 0x6f,
 	0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d,
-	0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x36, 0x0a, 0x06, 0x72, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x12,
-	0x12, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74,
-	0x69, 0x6f, 0x6e, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x35, 0x0a,
-	0x08, 0x67, 0x65, 0x74, 0x53, 0x74, 0x61, 0x74, 0x65, 0x12, 0x11, 0x2e, 0x63, 0x6f, 0x6e, 0x74,
-	0x65, 0x78, 0x74, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x1a, 0x14, 0x2e, 0x63,
-	0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x65, 0x52, 0x65, 0x73, 0x75,
-	0x6c, 0x74, 0x22, 0x00, 0x12, 0x3c, 0x0a, 0x08, 0x70, 0x75, 0x74, 0x53, 0x74, 0x61, 0x74, 0x65,
-	0x12, 0x16, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x65,
-	0x4b, 0x65, 0x79, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c,
-	0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79,
-	0x22, 0x00, 0x12, 0x3a, 0x0a, 0x0b, 0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74,
-	0x65, 0x12, 0x11, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x74, 0x61, 0x74,
-	0x65, 0x4b, 0x65, 0x79, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x33,
-	0x0a, 0x0a, 0x67, 0x65, 0x74, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x65, 0x72, 0x12, 0x11, 0x2e, 0x63,
-	0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x1a,
-	0x10, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x65,
-	0x72, 0x22, 0x00, 0x12, 0x3e, 0x0a, 0x0b, 0x69, 0x6e, 0x63, 0x72, 0x43, 0x6f, 0x75, 0x6e, 0x74,
-	0x65, 0x72, 0x12, 0x15, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x49, 0x6e, 0x63,
-	0x72, 0x53, 0x74, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67,
-	0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74,
-	0x79, 0x22, 0x00, 0x42, 0x06, 0x5a, 0x04, 0x2e, 0x3b, 0x70, 0x66, 0x62, 0x06, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x33,
+	0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x35, 0x0a, 0x05, 0x70, 0x61, 0x75, 0x73, 0x65, 0x12, 0x12,
+	0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69,
+	0x6f, 0x6e, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x36, 0x0a, 0x06,
+	0x72, 0x65, 0x73, 0x75, 0x6d, 0x65, 0x12, 0x12, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74,
+	0x2e, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f,
+	0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70,
+	0x74, 0x79, 0x22, 0x00, 0x12, 0x35, 0x0a, 0x08, 0x67, 0x65, 0x74, 0x53, 0x74, 0x61, 0x74, 0x65,
+	0x12, 0x11, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x65,
+	0x4b, 0x65, 0x79, 0x1a, 0x14, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x74,
+	0x61, 0x74, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x22, 0x00, 0x12, 0x3c, 0x0a, 0x08, 0x70,
+	0x75, 0x74, 0x53, 0x74, 0x61, 0x74, 0x65, 0x12, 0x16, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78,
+	0x74, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x1a,
+	0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75,
+	0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x0b, 0x64, 0x65, 0x6c,
+	0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x65, 0x12, 0x11, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65,
+	0x78, 0x74, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x1a, 0x16, 0x2e, 0x67, 0x6f,
+	0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d,
+	0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x33, 0x0a, 0x0a, 0x67, 0x65, 0x74, 0x43, 0x6f, 0x75, 0x6e,
+	0x74, 0x65, 0x72, 0x12, 0x11, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x2e, 0x53, 0x74,
+	0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x1a, 0x10, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74,
+	0x2e, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x65, 0x72, 0x22, 0x00, 0x12, 0x3e, 0x0a, 0x0b, 0x69, 0x6e,
+	0x63, 0x72, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x65, 0x72, 0x12, 0x15, 0x2e, 0x63, 0x6f, 0x6e, 0x74,
+	0x65, 0x78, 0x74, 0x2e, 0x49, 0x6e, 0x63, 0x72, 0x53, 0x74, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79,
+	0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
+	0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x42, 0x06, 0x5a, 0x04, 0x2e, 0x3b,
+	0x70, 0x66, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -940,7 +1046,7 @@ func file_context_proto_rawDescGZIP() []byte {
 	return file_context_proto_rawDescData
 }
 
-var file_context_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
+var file_context_proto_msgTypes = make([]protoimpl.MessageInfo, 14)
 var file_context_proto_goTypes = []interface{}{
 	(*SendMessageId)(nil), // 0: context.SendMessageId
 	(*StateKey)(nil),      // 1: context.StateKey
@@ -950,40 +1056,44 @@ var file_context_proto_goTypes = []interface{}{
 	(*Counter)(nil),       // 5: context.Counter
 	(*Request)(nil),       // 6: context.Request
 	(*MessageId)(nil),     // 7: context.MessageId
-	(*PulsarMessage)(nil), // 8: context.PulsarMessage
-	(*Record)(nil),        // 9: context.Record
-	(*MetricData)(nil),    // 10: context.MetricData
-	(*Partition)(nil),     // 11: context.Partition
-	(*emptypb.Empty)(nil), // 12: google.protobuf.Empty
+	(*PublishSchema)(nil), // 8: context.PublishSchema
+	(*PulsarMessage)(nil), // 9: context.PulsarMessage
+	(*Record)(nil),        // 10: context.Record
+	(*MetricData)(nil),    // 11: context.MetricData
+	(*Partition)(nil),     // 12: context.Partition
+	nil,                   // 13: context.PublishSchema.PropertiesEntry
+	(*emptypb.Empty)(nil), // 14: google.protobuf.Empty
 }
 var file_context_proto_depIdxs = []int32{
-	8,  // 0: context.ContextService.publish:input_type -> context.PulsarMessage
-	7,  // 1: context.ContextService.currentRecord:input_type -> context.MessageId
-	10, // 2: context.ContextService.recordMetrics:input_type -> context.MetricData
-	11, // 3: context.ContextService.seek:input_type -> context.Partition
-	11, // 4: context.ContextService.pause:input_type -> context.Partition
-	11, // 5: context.ContextService.resume:input_type -> context.Partition
-	1,  // 6: context.ContextService.getState:input_type -> context.StateKey
-	2,  // 7: context.ContextService.putState:input_type -> context.StateKeyValue
-	1,  // 8: context.ContextService.deleteState:input_type -> context.StateKey
-	1,  // 9: context.ContextService.getCounter:input_type -> context.StateKey
-	4,  // 10: context.ContextService.incrCounter:input_type -> context.IncrStateKey
-	0,  // 11: context.ContextService.publish:output_type -> context.SendMessageId
-	9,  // 12: context.ContextService.currentRecord:output_type -> context.Record
-	12, // 13: context.ContextService.recordMetrics:output_type -> google.protobuf.Empty
-	12, // 14: context.ContextService.seek:output_type -> google.protobuf.Empty
-	12, // 15: context.ContextService.pause:output_type -> google.protobuf.Empty
-	12, // 16: context.ContextService.resume:output_type -> google.protobuf.Empty
-	3,  // 17: context.ContextService.getState:output_type -> context.StateResult
-	12, // 18: context.ContextService.putState:output_type -> google.protobuf.Empty
-	12, // 19: context.ContextService.deleteState:output_type -> google.protobuf.Empty
-	5,  // 20: context.ContextService.getCounter:output_type -> context.Counter
-	12, // 21: context.ContextService.incrCounter:output_type -> google.protobuf.Empty
-	11, // [11:22] is the sub-list for method output_type
-	0,  // [0:11] is the sub-list for method input_type
-	0,  // [0:0] is the sub-list for extension type_name
-	0,  // [0:0] is the sub-list for extension extendee
-	0,  // [0:0] is the sub-list for field type_name
+	13, // 0: context.PublishSchema.properties:type_name -> context.PublishSchema.PropertiesEntry
+	8,  // 1: context.PulsarMessage.schema:type_name -> context.PublishSchema
+	9,  // 2: context.ContextService.publish:input_type -> context.PulsarMessage
+	7,  // 3: context.ContextService.currentRecord:input_type -> context.MessageId
+	11, // 4: context.ContextService.recordMetrics:input_type -> context.MetricData
+	12, // 5: context.ContextService.seek:input_type -> context.Partition
+	12, // 6: context.ContextService.pause:input_type -> context.Partition
+	12, // 7: context.ContextService.resume:input_type -> context.Partition
+	1,  // 8: context.ContextService.getState:input_type -> context.StateKey
+	2,  // 9: context.ContextService.putState:input_type -> context.StateKeyValue
+	1,  // 10: context.ContextService.deleteState:input_type -> context.StateKey
+	1,  // 11: context.ContextService.getCounter:input_type -> context.StateKey
+	4,  // 12: context.ContextService.incrCounter:input_type -> context.IncrStateKey
+	0,  // 13: context.ContextService.publish:output_type -> context.SendMessageId
+	10, // 14: context.ContextService.currentRecord:output_type -> context.Record
+	14, // 15: context.ContextService.recordMetrics:output_type -> google.protobuf.Empty
+	14, // 16: context.ContextService.seek:output_type -> google.protobuf.Empty
+	14, // 17: context.ContextService.pause:output_type -> google.protobuf.Empty
+	14, // 18: context.ContextService.resume:output_type -> google.protobuf.Empty
+	3,  // 19: context.ContextService.getState:output_type -> context.StateResult
+	14, // 20: context.ContextService.putState:output_type -> google.protobuf.Empty
+	14, // 21: context.ContextService.deleteState:output_type -> google.protobuf.Empty
+	5,  // 22: context.ContextService.getCounter:output_type -> context.Counter
+	14, // 23: context.ContextService.incrCounter:output_type -> google.protobuf.Empty
+	13, // [13:24] is the sub-list for method output_type
+	2,  // [2:13] is the sub-list for method input_type
+	2,  // [2:2] is the sub-list for extension type_name
+	2,  // [2:2] is the sub-list for extension extendee
+	0,  // [0:2] is the sub-list for field type_name
 }
 
 func init() { file_context_proto_init() }
@@ -1089,7 +1199,7 @@ func file_context_proto_init() {
 			}
 		}
 		file_context_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*PulsarMessage); i {
+			switch v := v.(*PublishSchema); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -1101,7 +1211,7 @@ func file_context_proto_init() {
 			}
 		}
 		file_context_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Record); i {
+			switch v := v.(*PulsarMessage); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -1113,7 +1223,7 @@ func file_context_proto_init() {
 			}
 		}
 		file_context_proto_msgTypes[10].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*MetricData); i {
+			switch v := v.(*Record); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -1125,6 +1235,18 @@ func file_context_proto_init() {
 			}
 		}
 		file_context_proto_msgTypes[11].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*MetricData); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_context_proto_msgTypes[12].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Partition); i {
 			case 0:
 				return &v.state
@@ -1143,7 +1265,7 @@ func file_context_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_context_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   12,
+			NumMessages:   14,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/pf/context_test.go b/pf/context_test.go
index 7c67076..18854dd 100644
--- a/pf/context_test.go
+++ b/pf/context_test.go
@@ -1,6 +1,12 @@
 package pf
 
-import "testing"
+import (
+	"context"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
 
 func TestRecordEventTimestampIsInt64(t *testing.T) {
 	record := &Record{EventTimestamp: 5}
@@ -22,3 +28,174 @@ func TestSetMessageIdClearsCachedRecord(t *testing.T) {
 		t.Fatalf("cached message = %+v, want nil", functionContext.message)
 	}
 }
+
+func TestPublishSendsRawPayloadWithoutSchema(t *testing.T) {
+	stub := &publishCaptureStub{}
+	functionContext := &FunctionContext{
+		ctx:  context.Background(),
+		stub: stub,
+	}
+
+	messageID, err := functionContext.Publish("persistent://public/default/raw", []byte("payload"))
+	if err != nil {
+		t.Fatalf("Publish returned error: %v", err)
+	}
+
+	if messageID.GetEntryId() != 7 {
+		t.Fatalf("entryId = %d, want 7", messageID.GetEntryId())
+	}
+	if stub.message == nil {
+		t.Fatal("published message is nil")
+	}
+	if stub.message.GetTopic() != "persistent://public/default/raw" {
+		t.Fatalf("topic = %q, want raw topic", stub.message.GetTopic())
+	}
+	if string(stub.message.GetPayload()) != "payload" {
+		t.Fatalf("payload = %q, want payload", stub.message.GetPayload())
+	}
+	if stub.message.GetSchema() != nil {
+		t.Fatalf("schema = %+v, want nil", stub.message.GetSchema())
+	}
+}
+
+func TestPublishWithSchemaSendsSchemaMetadata(t *testing.T) {
+	stub := &publishCaptureStub{}
+	functionContext := &FunctionContext{
+		ctx:  context.Background(),
+		stub: stub,
+	}
+	schema := PublishMessageSchema{
+		SchemaType: SchemaTypeAvro,
+		Name:       "example.Event",
+		SchemaData: `{"type":"record","name":"Event","fields":[]}`,
+		Properties: map[string]string{
+			"encoding": "binary",
+		},
+		Subject: "events-value",
+	}
+
+	_, err := functionContext.PublishWithSchema(
+		"persistent://public/default/events",
+		[]byte("encoded"),
+		schema,
+	)
+	if err != nil {
+		t.Fatalf("PublishWithSchema returned error: %v", err)
+	}
+
+	got := stub.message.GetSchema()
+	if got == nil {
+		t.Fatal("schema is nil, want schema metadata")
+	}
+	if got.GetSchemaType() != SchemaTypeAvro {
+		t.Fatalf("schemaType = %q, want %q", got.GetSchemaType(), SchemaTypeAvro)
+	}
+	if got.GetName() != "example.Event" {
+		t.Fatalf("schema name = %q, want example.Event", got.GetName())
+	}
+	if string(got.GetSchemaData()) != `{"type":"record","name":"Event","fields":[]}` {
+		t.Fatalf("schemaData = %q, want Avro definition", got.GetSchemaData())
+	}
+	if got.GetProperties()["encoding"] != "binary" {
+		t.Fatalf("schema properties = %+v, want encoding=binary", got.GetProperties())
+	}
+	if got.GetSubject() != "events-value" {
+		t.Fatalf("subject = %q, want events-value", got.GetSubject())
+	}
+	if string(stub.message.GetPayload()) != "encoded" {
+		t.Fatalf("payload = %q, want encoded", stub.message.GetPayload())
+	}
+}
+
+func TestPublishWithSchemaNoSchemaTypesPublishRawPayload(t *testing.T) {
+	tests := []struct {
+		name       string
+		schemaType string
+	}{
+		{name: "empty", schemaType: ""},
+		{name: "none", schemaType: SchemaTypeNone},
+		{name: "bytes", schemaType: SchemaTypeBytes},
+		{name: "trimmed uppercase bytes", schemaType: " BYTES "},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			stub := &publishCaptureStub{}
+			functionContext := &FunctionContext{
+				ctx:  context.Background(),
+				stub: stub,
+			}
+
+			_, err := functionContext.PublishWithSchema(
+				"persistent://public/default/raw",
+				[]byte("payload"),
+				PublishMessageSchema{
+					SchemaType: tt.schemaType,
+					Name:       "ignored",
+					SchemaData: `{"type":"record","name":"Ignored","fields":[]}`,
+				},
+			)
+			if err != nil {
+				t.Fatalf("PublishWithSchema returned error: %v", err)
+			}
+			if stub.message == nil {
+				t.Fatal("published message is nil")
+			}
+			if string(stub.message.GetPayload()) != "payload" {
+				t.Fatalf("payload = %q, want payload", stub.message.GetPayload())
+			}
+			if stub.message.GetSchema() != nil {
+				t.Fatalf("schema = %+v, want nil", stub.message.GetSchema())
+			}
+		})
+	}
+}
+
+type publishCaptureStub struct {
+	message *PulsarMessage
+}
+
+func (s *publishCaptureStub) Publish(_ context.Context, message *PulsarMessage, _ ...grpc.CallOption) (*SendMessageId, error) {
+	s.message = message
+	return &SendMessageId{EntryId: 7}, nil
+}
+
+func (s *publishCaptureStub) CurrentRecord(context.Context, *MessageId, ...grpc.CallOption) (*Record, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) RecordMetrics(context.Context, *MetricData, ...grpc.CallOption) (*emptypb.Empty, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) Seek(context.Context, *Partition, ...grpc.CallOption) (*emptypb.Empty, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) Pause(context.Context, *Partition, ...grpc.CallOption) (*emptypb.Empty, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) Resume(context.Context, *Partition, ...grpc.CallOption) (*emptypb.Empty, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) GetState(context.Context, *StateKey, ...grpc.CallOption) (*StateResult, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) PutState(context.Context, *StateKeyValue, ...grpc.CallOption) (*emptypb.Empty, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) DeleteState(context.Context, *StateKey, ...grpc.CallOption) (*emptypb.Empty, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) GetCounter(context.Context, *StateKey, ...grpc.CallOption) (*Counter, error) {
+	return nil, nil
+}
+
+func (s *publishCaptureStub) IncrCounter(context.Context, *IncrStateKey, ...grpc.CallOption) (*emptypb.Empty, error) {
+	return nil, nil
+}