Repository: apache/pulsar-client-go
Pull request #1006 (2023-04-10 13:41:45)

[feat][schema] Support the schema type ProtoNativeSchema
### Motivation

The pulsar-client-go supports ProtoSchema based on the Avro format, users must provide the schema definition string when using struct schema(JSON, Avro, Proto), it's hard to use and we can't get the entire original proto definition. 

The difference between ProtoSchema and ProtoNativeSchema is the schema data, the ProtoNativeSchema will contain the entire original proto definition, and we can use the schema data of the ProtoNativeSchema to generate the corresponding Descriptor and get any field, we can leverage this feature in Pulsar SQL.

The schema data of the ProtoNativeSchema follow the Pulsar [format](https://github.com/apache/pulsar/blob/master/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/ProtobufNativeSchemaData.java), we can generate JSON data for `ProtobufNativeSchemaData` as the schema data.

```
public class ProtobufNativeSchemaData {

    private byte[] fileDescriptorSet;
    private String rootMessageTypeName;
    private String rootFileDescriptorName;

}
```

The field `fileDescriptorSet` is the serialized data of the message `FileDescriptorSet`, it contains all definition proto files data.

```
message FileDescriptorSet {
  repeated FileDescriptorProto file = 1;
}
```

```
message FileDescriptorProto {
  optional string name = 1;
  optional string package = 2;
  repeated string dependency = 3;
  repeated DescriptorProto message_type = 4;
  repeated EnumDescriptorProto enum_type = 5;
  repeated ServiceDescriptorProto service = 6;
  repeated FieldDescriptorProto extension = 7;
  optional FileOptions options = 8;
  optional SourceCodeInfo source_code_info = 9;
  repeated int32 public_dependency = 10;
  repeated int32 weak_dependency = 11;
  optional string syntax = 12;
} 
```

### Modifications

Add a new schema type `ProtoNative(20)`.
Add ProtoNativeSchema.
Add encode and decode methods for ProtoNativeSchema.

### Verifying this change

Add an integration test to make sure the broker can deserialize the schema data of the ProtoNativeSchema.

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (yes)
  - The public API: (yes)
  - The schema: (yes)
  - The default values of configurations: (no)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation


BewareMyPower: Could you fix lint error? You can use `golangci-lint run` to check lint errors in your local env.
RobertIndie: https://github.com/apache/pulsar-client-go/actions/runs/4670464787/jobs/8270224455?pr=1006#step:5:789

Patch:
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
index 343963e..4641eb9 100644
--- a/.github/workflows/bot.yml
+++ b/.github/workflows/bot.yml
@@ -29,6 +29,6 @@ jobs:
       - name: Execute pulsarbot command
         id: pulsarbot
         env:
-          GITHUB_TOKEN: ${{ secrets.GO_CLIENT_BOT_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           BOT_TARGET_REPOSITORY: apache/pulsar-client-go
         uses: apache/pulsar-test-infra/pulsarbot@master
diff --git a/go.mod b/go.mod
index 3e7299b..69862b2 100644
--- a/go.mod
+++ b/go.mod
@@ -27,7 +27,7 @@ require (
 	golang.org/x/mod v0.5.1
 	golang.org/x/oauth2 v0.0.0-20210402161424-2e8d93401602
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
-	google.golang.org/protobuf v1.26.0
+	google.golang.org/protobuf v1.30.0 // indirect
 )
 
 require (
diff --git a/go.sum b/go.sum
index 528c458..3a39874 100644
--- a/go.sum
+++ b/go.sum
@@ -553,6 +553,8 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0 h1:bxAC2xTBsZGibn2RTntX0oH50xLsqy1OxA9tTL3p/lk=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index fb77d0d..a8ba7b5 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -1525,6 +1525,11 @@ func (pc *partitionConsumer) reconnectToBroker() {
 		maxRetry = int(*pc.options.maxReconnectToBroker)
 	}
 
+	var (
+		delayReconnectTime time.Duration
+		defaultBackoff     = internal.DefaultBackoff{}
+	)
+
 	for maxRetry != 0 {
 		if pc.getConsumerState() != consumerReady {
 			// Consumer is already closing
@@ -1532,11 +1537,6 @@ func (pc *partitionConsumer) reconnectToBroker() {
 			return
 		}
 
-		var (
-			delayReconnectTime time.Duration
-			defaultBackoff     = internal.DefaultBackoff{}
-		)
-
 		if pc.options.backoffPolicy == nil {
 			delayReconnectTime = defaultBackoff.Next()
 		} else {
diff --git a/pulsar/schema.go b/pulsar/schema.go
index 997d85f..405049d 100644
--- a/pulsar/schema.go
+++ b/pulsar/schema.go
@@ -30,6 +30,9 @@ import (
 
 	"github.com/linkedin/goavro/v2"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protodesc"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/types/descriptorpb"
 )
 
 type SchemaType int
@@ -55,6 +58,7 @@ const (
 	AUTO        = -2              //
 	AutoConsume = -3              //Auto Consume Type.
 	AutoPublish = -4              // Auto Publish Type.
+	ProtoNative = 20              //Protobuf native message encoding and decoding
 )
 
 // Encapsulates data around the schema definition
@@ -102,6 +106,8 @@ func NewSchema(schemaType SchemaType, schemaData []byte, properties map[string]s
 		s = NewFloatSchema(properties)
 	case DOUBLE:
 		s = NewDoubleSchema(properties)
+	case ProtoNative:
+		s = newProtoNativeSchema(schemaDef, properties)
 	default:
 		err = fmt.Errorf("not support schema type of %v", schemaType)
 	}
@@ -219,6 +225,84 @@ func (ps *ProtoSchema) GetSchemaInfo() *SchemaInfo {
 	return &ps.SchemaInfo
 }
 
+type ProtoNativeSchema struct {
+	SchemaInfo
+}
+
+func NewProtoNativeSchemaWithMessage(message proto.Message, properties map[string]string) *ProtoNativeSchema {
+	schemaDef, err := getProtoNativeSchemaInfo(message)
+	if err != nil {
+		log.Fatalf("Get ProtoNative schema info error:%v", err)
+	}
+	return newProtoNativeSchema(schemaDef, properties)
+}
+
+func newProtoNativeSchema(protoNativeSchemaDef string, properties map[string]string) *ProtoNativeSchema {
+	pns := new(ProtoNativeSchema)
+	pns.SchemaInfo.Schema = protoNativeSchemaDef
+	pns.SchemaInfo.Type = ProtoNative
+	pns.SchemaInfo.Properties = properties
+	pns.SchemaInfo.Name = "ProtoNative"
+	return pns
+}
+
+func getProtoNativeSchemaInfo(message proto.Message) (string, error) {
+	fileDesc := message.ProtoReflect().Descriptor().ParentFile()
+	fileProtoMap := make(map[string]*descriptorpb.FileDescriptorProto)
+	getFileProto(fileDesc, fileProtoMap)
+
+	fileDescList := make([]*descriptorpb.FileDescriptorProto, 0, len(fileProtoMap))
+	for _, v := range fileProtoMap {
+		fileDescList = append(fileDescList, v)
+	}
+	fileDescSet := descriptorpb.FileDescriptorSet{
+		File: fileDescList,
+	}
+	bytesData, err := proto.Marshal(&fileDescSet)
+	if err != nil {
+		return "", err
+	}
+	schemaData := ProtoNativeSchemaData{
+		FileDescriptorSet:      bytesData,
+		RootMessageTypeName:    string(message.ProtoReflect().Descriptor().FullName()),
+		RootFileDescriptorName: fileDesc.Path(),
+	}
+	jsonData, err := json.Marshal(schemaData)
+	if err != nil {
+		return "", err
+	}
+	return string(jsonData), nil
+}
+
+type ProtoNativeSchemaData struct {
+	FileDescriptorSet      []byte `json:"fileDescriptorSet"`
+	RootMessageTypeName    string `json:"rootMessageTypeName"`
+	RootFileDescriptorName string `json:"rootFileDescriptorName"`
+}
+
+func getFileProto(fileDesc protoreflect.FileDescriptor, protoMap map[string]*descriptorpb.FileDescriptorProto) {
+	for i := 0; i < fileDesc.Imports().Len(); i++ {
+		getFileProto(fileDesc.Imports().Get(i).ParentFile(), protoMap)
+	}
+	protoMap[fileDesc.Path()] = protodesc.ToFileDescriptorProto(fileDesc)
+}
+
+func (ps *ProtoNativeSchema) Encode(data interface{}) ([]byte, error) {
+	return proto.Marshal(data.(proto.Message))
+}
+
+func (ps *ProtoNativeSchema) Decode(data []byte, v interface{}) error {
+	return proto.Unmarshal(data, v.(proto.Message))
+}
+
+func (ps *ProtoNativeSchema) Validate(message []byte) error {
+	return ps.Decode(message, nil)
+}
+
+func (ps *ProtoNativeSchema) GetSchemaInfo() *SchemaInfo {
+	return &ps.SchemaInfo
+}
+
 type AvroSchema struct {
 	AvroCodec
 	SchemaInfo


Test patch:
diff --git a/integration-tests/pb/build.sh b/integration-tests/pb/build.sh
index 32c0b08..e192458 100755
--- a/integration-tests/pb/build.sh
+++ b/integration-tests/pb/build.sh
@@ -18,4 +18,9 @@
 # under the License.
 #
 
-protoc --go_out=. --go_opt=paths=source_relative hello.proto
+PROJECT_DIR=$(git rev-parse --show-toplevel)
+SOURCE_PATH="$PROJECT_DIR/integration-tests/pb"
+DEST_PATH="$PROJECT_DIR/integration-tests/pb"
+echo "source_path: $SOURCE_PATH"
+echo "dest_path: $DEST_PATH"
+protoc -I=$SOURCE_PATH --go_out=$DEST_PATH $SOURCE_PATH/*.proto
diff --git a/integration-tests/pb/hello.pb.go b/integration-tests/pb/hello.pb.go
index 7884dfa..541172c 100644
--- a/integration-tests/pb/hello.pb.go
+++ b/integration-tests/pb/hello.pb.go
@@ -18,11 +18,11 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.28.1
-// 	protoc        v3.21.9
+// 	protoc-gen-go v1.30.0
+// 	protoc        v3.21.12
 // source: hello.proto
 
-package pb
+package __
 
 import (
 	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
@@ -43,8 +43,10 @@ type Test struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Num int32  `protobuf:"varint,1,opt,name=num,proto3" json:"num,omitempty"`
-	Msf string `protobuf:"bytes,2,opt,name=msf,proto3" json:"msf,omitempty"`
+	Num       int32      `protobuf:"varint,1,opt,name=num,proto3" json:"num,omitempty"`
+	Msf       string     `protobuf:"bytes,2,opt,name=msf,proto3" json:"msf,omitempty"`
+	Foo       *Foo       `protobuf:"bytes,3,opt,name=foo,proto3" json:"foo,omitempty"`
+	HiContent *HiContent `protobuf:"bytes,4,opt,name=hiContent,proto3" json:"hiContent,omitempty"`
 }
 
 func (x *Test) Reset() {
@@ -93,15 +95,102 @@ func (x *Test) GetMsf() string {
 	return ""
 }
 
+func (x *Test) GetFoo() *Foo {
+	if x != nil {
+		return x.Foo
+	}
+	return nil
+}
+
+func (x *Test) GetHiContent() *HiContent {
+	if x != nil {
+		return x.HiContent
+	}
+	return nil
+}
+
+type Foo struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Name  string   `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	Value int32    `protobuf:"varint,2,opt,name=value,proto3" json:"value,omitempty"`
+	Names []string `protobuf:"bytes,3,rep,name=names,proto3" json:"names,omitempty"`
+}
+
+func (x *Foo) Reset() {
+	*x = Foo{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_hello_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *Foo) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Foo) ProtoMessage() {}
+
+func (x *Foo) ProtoReflect() protoreflect.Message {
+	mi := &file_hello_proto_msgTypes[1]
+	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 Foo.ProtoReflect.Descriptor instead.
+func (*Foo) Descriptor() ([]byte, []int) {
+	return file_hello_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *Foo) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *Foo) GetValue() int32 {
+	if x != nil {
+		return x.Value
+	}
+	return 0
+}
+
+func (x *Foo) GetNames() []string {
+	if x != nil {
+		return x.Names
+	}
+	return nil
+}
+
 var File_hello_proto protoreflect.FileDescriptor
 
 var file_hello_proto_rawDesc = []byte{
 	0x0a, 0x0b, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74, 0x22, 0x2a, 0x0a, 0x04, 0x54, 0x65, 0x73, 0x74,
-	0x12, 0x10, 0x0a, 0x03, 0x6e, 0x75, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6e,
-	0x75, 0x6d, 0x12, 0x10, 0x0a, 0x03, 0x6d, 0x73, 0x66, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x03, 0x6d, 0x73, 0x66, 0x42, 0x06, 0x5a, 0x04, 0x2e, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74, 0x1a, 0x08, 0x68, 0x69, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x22, 0x80, 0x01, 0x0a, 0x04, 0x54, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x6e,
+	0x75, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6e, 0x75, 0x6d, 0x12, 0x10, 0x0a,
+	0x03, 0x6d, 0x73, 0x66, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6d, 0x73, 0x66, 0x12,
+	0x20, 0x0a, 0x03, 0x66, 0x6f, 0x6f, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74, 0x2e, 0x46, 0x6f, 0x6f, 0x52, 0x03, 0x66, 0x6f,
+	0x6f, 0x12, 0x32, 0x0a, 0x09, 0x68, 0x69, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x18, 0x04,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x74, 0x65, 0x73, 0x74,
+	0x2e, 0x48, 0x69, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x52, 0x09, 0x68, 0x69, 0x43, 0x6f,
+	0x6e, 0x74, 0x65, 0x6e, 0x74, 0x22, 0x45, 0x0a, 0x03, 0x46, 0x6f, 0x6f, 0x12, 0x12, 0x0a, 0x04,
+	0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65,
+	0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52,
+	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x18,
+	0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x05, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x42, 0x03, 0x5a, 0x01,
+	0x2e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -116,16 +205,20 @@ func file_hello_proto_rawDescGZIP() []byte {
 	return file_hello_proto_rawDescData
 }
 
-var file_hello_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_hello_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
 var file_hello_proto_goTypes = []interface{}{
-	(*Test)(nil), // 0: prototest.Test
+	(*Test)(nil),      // 0: prototest.Test
+	(*Foo)(nil),       // 1: prototest.Foo
+	(*HiContent)(nil), // 2: prototest.HiContent
 }
 var file_hello_proto_depIdxs = []int32{
-	0, // [0:0] is the sub-list for method output_type
-	0, // [0:0] 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
+	1, // 0: prototest.Test.foo:type_name -> prototest.Foo
+	2, // 1: prototest.Test.hiContent:type_name -> prototest.HiContent
+	2, // [2:2] is the sub-list for method output_type
+	2, // [2:2] 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_hello_proto_init() }
@@ -133,6 +226,7 @@ func file_hello_proto_init() {
 	if File_hello_proto != nil {
 		return
 	}
+	file_hi_proto_init()
 	if !protoimpl.UnsafeEnabled {
 		file_hello_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Test); i {
@@ -146,6 +240,18 @@ func file_hello_proto_init() {
 				return nil
 			}
 		}
+		file_hello_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Foo); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -153,7 +259,7 @@ func file_hello_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_hello_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   1,
+			NumMessages:   2,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/integration-tests/pb/hello.proto b/integration-tests/pb/hello.proto
index 3a67c3b..ca6a4ff 100644
--- a/integration-tests/pb/hello.proto
+++ b/integration-tests/pb/hello.proto
@@ -19,9 +19,19 @@
 syntax = "proto3";
 package prototest;
 
-option go_package = "./pb";
+option go_package = ".";
+
+import "hi.proto";
 
 message Test {
     int32 num = 1;
     string msf = 2;
+    Foo foo = 3;
+    HiContent hiContent = 4;
+}
+
+message Foo {
+    string name = 1;
+    int32 value = 2;
+    repeated string names = 3;
 }
diff --git a/integration-tests/pb/hi.pb.go b/integration-tests/pb/hi.pb.go
new file mode 100644
index 0000000..1027caf
--- /dev/null
+++ b/integration-tests/pb/hi.pb.go
@@ -0,0 +1,168 @@
+//*
+// 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.
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.30.0
+// 	protoc        v3.21.12
+// source: hi.proto
+
+package __
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type HiContent struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Id      int32  `protobuf:"varint,2,opt,name=id,proto3" json:"id,omitempty"`
+	Content string `protobuf:"bytes,1,opt,name=content,proto3" json:"content,omitempty"`
+}
+
+func (x *HiContent) Reset() {
+	*x = HiContent{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_hi_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *HiContent) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*HiContent) ProtoMessage() {}
+
+func (x *HiContent) ProtoReflect() protoreflect.Message {
+	mi := &file_hi_proto_msgTypes[0]
+	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 HiContent.ProtoReflect.Descriptor instead.
+func (*HiContent) Descriptor() ([]byte, []int) {
+	return file_hi_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *HiContent) GetId() int32 {
+	if x != nil {
+		return x.Id
+	}
+	return 0
+}
+
+func (x *HiContent) GetContent() string {
+	if x != nil {
+		return x.Content
+	}
+	return ""
+}
+
+var File_hi_proto protoreflect.FileDescriptor
+
+var file_hi_proto_rawDesc = []byte{
+	0x0a, 0x08, 0x68, 0x69, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x74, 0x65, 0x73, 0x74, 0x22, 0x35, 0x0a, 0x09, 0x48, 0x69, 0x43, 0x6f, 0x6e, 0x74, 0x65,
+	0x6e, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x02,
+	0x69, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x42, 0x03, 0x5a, 0x01,
+	0x2e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_hi_proto_rawDescOnce sync.Once
+	file_hi_proto_rawDescData = file_hi_proto_rawDesc
+)
+
+func file_hi_proto_rawDescGZIP() []byte {
+	file_hi_proto_rawDescOnce.Do(func() {
+		file_hi_proto_rawDescData = protoimpl.X.CompressGZIP(file_hi_proto_rawDescData)
+	})
+	return file_hi_proto_rawDescData
+}
+
+var file_hi_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_hi_proto_goTypes = []interface{}{
+	(*HiContent)(nil), // 0: prototest.HiContent
+}
+var file_hi_proto_depIdxs = []int32{
+	0, // [0:0] is the sub-list for method output_type
+	0, // [0:0] 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
+}
+
+func init() { file_hi_proto_init() }
+func file_hi_proto_init() {
+	if File_hi_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_hi_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*HiContent); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_hi_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_hi_proto_goTypes,
+		DependencyIndexes: file_hi_proto_depIdxs,
+		MessageInfos:      file_hi_proto_msgTypes,
+	}.Build()
+	File_hi_proto = out.File
+	file_hi_proto_rawDesc = nil
+	file_hi_proto_goTypes = nil
+	file_hi_proto_depIdxs = nil
+}
diff --git a/integration-tests/pb/hi.proto b/integration-tests/pb/hi.proto
new file mode 100644
index 0000000..f5a401a
--- /dev/null
+++ b/integration-tests/pb/hi.proto
@@ -0,0 +1,9 @@
+syntax = "proto3";
+package prototest;
+
+option go_package = ".";
+
+message HiContent {
+  int32 id = 2;
+  string content = 1;
+}
diff --git a/pulsar/schema_test.go b/pulsar/schema_test.go
index e7d52ca..c2008f6 100644
--- a/pulsar/schema_test.go
+++ b/pulsar/schema_test.go
@@ -22,7 +22,7 @@ import (
 	"log"
 	"testing"
 
-	"github.com/apache/pulsar-client-go/integration-tests/pb"
+	pb "github.com/apache/pulsar-client-go/integration-tests/pb"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -147,6 +147,69 @@ func TestProtoSchema(t *testing.T) {
 	defer consumer.Close()
 }
 
+func TestProtoNativeSchema(t *testing.T) {
+	client := createClient()
+	defer client.Close()
+
+	topic := "proto-native"
+
+	// create producer
+	psProducer := NewProtoNativeSchemaWithMessage(&pb.Test{}, nil)
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  topic,
+		Schema: psProducer,
+	})
+	assert.Nil(t, err)
+
+	names := []string{
+		"name-a",
+		"name-b",
+		"name-c",
+	}
+	if _, err := producer.Send(context.Background(), &ProducerMessage{
+		Value: &pb.Test{
+			Num: 100,
+			Msf: "pulsar",
+			Foo: &pb.Foo{
+				Name:  "foo name",
+				Value: 200,
+				Names: names,
+			},
+			HiContent: &pb.HiContent{
+				Id:      300,
+				Content: "hi content",
+			},
+		},
+	}); err != nil {
+		log.Fatal(err)
+	}
+
+	//create consumer
+	unobj := pb.Test{}
+	psConsumer := NewProtoNativeSchemaWithMessage(&pb.Test{}, nil)
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       topic,
+		SubscriptionName:            "sub-1",
+		Schema:                      psConsumer,
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+	})
+	assert.Nil(t, err)
+
+	msg, err := consumer.Receive(context.Background())
+	assert.Nil(t, err)
+	err = msg.GetSchemaValue(&unobj)
+	assert.Nil(t, err)
+	assert.Equal(t, unobj.Num, int32(100))
+	assert.Equal(t, unobj.Msf, "pulsar")
+	assert.Equal(t, unobj.Foo.Name, "foo name")
+	assert.Equal(t, unobj.Foo.Value, int32(200))
+	assert.Equal(t, unobj.Foo.Names, names)
+	assert.Equal(t, unobj.HiContent.Id, int32(300))
+	assert.Equal(t, unobj.HiContent.Content, "hi content")
+	consumer.Ack(msg)
+	defer consumer.Close()
+}
+
 func TestAvroSchema(t *testing.T) {
 	client := createClient()
 	defer client.Close()