From c919588fc041bb683d7670834cb2b2a116494b5e Mon Sep 17 00:00:00 2001
From: wangguowei <wangguowei@zhihu.com>
Date: Mon, 26 Oct 2020 15:21:40 +0800
Subject: [PATCH 01/10] [WIP]ProtobufNative Schema Support

---
 .../pulsar/broker/ServiceConfiguration.java   |   9 +-
 ...rotobufNativeSchemaCompatibilityCheck.java |  72 +++++++++
 .../ProtobufNativeSchemaDataValidator.java    |  49 ++++++
 .../schema/validator/SchemaDataValidator.java |   3 +
 .../src/main/proto/SchemaRegistryFormat.proto |   3 +-
 .../org/apache/pulsar/client/api/Schema.java  |  21 +++
 .../internal/DefaultImplementation.java       |  28 +++-
 .../pulsar/common/schema/SchemaInfo.java      |   1 +
 .../pulsar/common/schema/SchemaType.java      |   7 +
 .../client/impl/schema/AutoConsumeSchema.java |  31 ++--
 .../impl/schema/ProtobufNativeSchema.java     | 146 ++++++++++++++++++
 .../schema/ProtobufNativeSchemaUtils.java     | 139 +++++++++++++++++
 .../generic/GenericProtobufNativeReader.java  |  83 ++++++++++
 .../generic/GenericProtobufNativeRecord.java  |  46 ++++++
 .../generic/GenericProtobufNativeSchema.java  | 104 +++++++++++++
 .../generic/GenericProtobufNativeWriter.java  |  29 ++++
 .../ProtobufNativeRecordBuilderImpl.java      |  72 +++++++++
 .../schema/reader/ProtobufNativeReader.java   |  27 ++++
 .../schema/writer/ProtobufNativeWriter.java   |  27 ++++
 .../pulsar/common/api/proto/PulsarApi.java    |   3 +
 .../schema/ProtobufNativeSchemaData.java      |  47 ++++++
 pulsar-common/src/main/proto/PulsarApi.proto  |   1 +
 22 files changed, 929 insertions(+), 19 deletions(-)
 create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
 create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReader.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeRecord.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeWriter.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/ProtobufNativeRecordBuilderImpl.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java
 create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java
 create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/ProtobufNativeSchemaData.java

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index c0fa3f9be1516..9be91ae52813c 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -239,7 +239,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @FieldContext(
         category = CATEGORY_SERVER,
         dynamic = true,
-        doc = "The maximum number of tenants that each pulsar cluster can create." 
+        doc = "The maximum number of tenants that each pulsar cluster can create."
                 + "This configuration is not precise control, in a concurrent scenario, the threshold will be exceeded."
     )
     private int maxTenants = 0;
@@ -430,9 +430,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "When a namespace is created without specifying the number of bundle, this"
             + " value will be used as the default")
     private int defaultNumberOfNamespaceBundles = 4;
-    
+
     @FieldContext(
-        category = CATEGORY_POLICIES, 
+        category = CATEGORY_POLICIES,
         dynamic = true,
         doc = "The maximum number of namespaces that each tenant can create."
             + "This configuration is not precise control, in a concurrent scenario, the threshold will be exceeded")
@@ -1689,7 +1689,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private Set<String> schemaRegistryCompatibilityCheckers = Sets.newHashSet(
             "org.apache.pulsar.broker.service.schema.JsonSchemaCompatibilityCheck",
-            "org.apache.pulsar.broker.service.schema.AvroSchemaCompatibilityCheck"
+            "org.apache.pulsar.broker.service.schema.AvroSchemaCompatibilityCheck",
+            "org.apache.pulsar.broker.service.schema.ProtobufNativeSchemaCompatibilityCheck"
     );
 
     /**** --- WebSocket --- ****/
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
new file mode 100644
index 0000000000000..e8d712bbc5253
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
@@ -0,0 +1,72 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaUtils;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+import static com.google.protobuf.Descriptors.Descriptor;
+
+/**
+ * The {@link SchemaCompatibilityCheck} implementation for {@link SchemaType#PROTOBUF_NATIVE}.
+ */
+public class ProtobufNativeSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+    @Override
+    public SchemaType getSchemaType() {
+        return SchemaType.PROTOBUF_NATIVE;
+    }
+
+    @Override
+    public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy) throws IncompatibleSchemaException {
+        Descriptor fromDescriptor = ProtobufNativeSchemaUtils.deserialize(from.getData());
+        Descriptor toDescriptor = ProtobufNativeSchemaUtils.deserialize(from.getData());
+        switch (strategy) {
+            case BACKWARD_TRANSITIVE:
+            case BACKWARD:
+            case FORWARD_TRANSITIVE:
+            case FORWARD:
+            case FULL_TRANSITIVE:
+            case FULL:
+                CheckRootRootMessageChange(fromDescriptor, toDescriptor, strategy);
+                return;
+            case ALWAYS_COMPATIBLE:
+                return;
+            default:
+                throw new IncompatibleSchemaException("Unknown SchemaCompatibilityStrategy");
+        }
+    }
+
+    @Override
+    public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaCompatibilityStrategy strategy) throws IncompatibleSchemaException {
+        for (SchemaData schemaData : from) {
+            checkCompatible(schemaData, to, strategy);
+        }
+    }
+
+    private void CheckRootRootMessageChange(Descriptor fromDescriptor, Descriptor toDescriptor, SchemaCompatibilityStrategy strategy) throws IncompatibleSchemaException {
+        if (!fromDescriptor.getFullName().equals(toDescriptor.getFullName())) {
+            throw new IncompatibleSchemaException("Protobuf root message isn't allow change!");
+        }
+    }
+
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
new file mode 100644
index 0000000000000..5479f4845cda6
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
@@ -0,0 +1,49 @@
+/**
+ * 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 org.apache.pulsar.broker.service.schema.validator;
+
+import com.google.protobuf.Descriptors;
+import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException;
+import org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaUtils;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+
+public class ProtobufNativeSchemaDataValidator implements SchemaDataValidator {
+
+    @Override
+    public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
+        Descriptors.Descriptor descriptor;
+        try {
+            descriptor = ProtobufNativeSchemaUtils.deserialize(schemaData.getData());
+        } catch (Exception e) {
+            throw new InvalidSchemaDataException("deserialize  ProtobufNative Schema failed", e);
+        }
+        if (descriptor == null) {
+            throw new InvalidSchemaDataException("protobuf root message Descriptor is null , please recheck rootMessageTypeName or rootFileDescriptorName conf. ");
+        }
+    }
+
+    public static ProtobufNativeSchemaDataValidator of() {
+        return INSTANCE;
+    }
+
+    private static final ProtobufNativeSchemaDataValidator INSTANCE = new ProtobufNativeSchemaDataValidator();
+
+    private ProtobufNativeSchemaDataValidator() {
+    }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidator.java
index 5e5f846a7ea11..34a5f3447cf80 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidator.java
@@ -41,6 +41,9 @@ static void validateSchemaData(SchemaData schemaData) throws InvalidSchemaDataEx
             case PROTOBUF:
                 StructSchemaDataValidator.of().validate(schemaData);
                 break;
+            case PROTOBUF_NATIVE:
+                ProtobufNativeSchemaDataValidator.of().validate(schemaData);
+                break;
             case STRING:
                 StringSchemaDataValidator.of().validate(schemaData);
                 break;
diff --git a/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto b/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto
index 32790b2868411..5bdec8b563b24 100644
--- a/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto
+++ b/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto
@@ -1,4 +1,4 @@
-/**
+   /**
  * 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
@@ -44,6 +44,7 @@ message SchemaInfo {
 		LOCALDATE = 18;
 		LOCALTIME = 19;
 		LOCALDATETIME = 20;
+		PROTOBUFNATIVE = 21;
     }
     message KeyValuePair {
         required string key = 1;
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
index e7e5bbe420ec8..36f099325a2a4 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
@@ -251,6 +251,27 @@ static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF(Sch
         return DefaultImplementation.newProtobufSchema(schemaDefinition);
     }
 
+    /**
+     * Create a Protobuf-Native schema type by extracting the fields of the specified class.
+     *
+     * @param clazz the Protobuf generated class to be used to extract the schema
+     * @return a Schema instance
+     */
+    static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUFNATIVE(Class<T> clazz) {
+        return DefaultImplementation.newProtobufNativeSchema(SchemaDefinition.builder().withPojo(clazz).build());
+    }
+
+    /**
+     * Create a Protobuf-Native schema type with schema definition.
+     *
+     * @param schemaDefinition schemaDefinition the definition of the schema
+     * @return a Schema instance
+     */
+    static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUFNATIVE(
+            SchemaDefinition<T> schemaDefinition) {
+        return DefaultImplementation.newProtobufNativeSchema(schemaDefinition);
+    }
+
     /**
      * Create a  Avro schema type by default configuration of the class.
      *
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
index 589ac69cd2b37..9760cad2e50a9 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
@@ -263,6 +263,14 @@ public static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newPr
                 .invoke(null, schemaDefinition));
     }
 
+    public static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufNativeSchema(
+            SchemaDefinition schemaDefinition) {
+        return catchExceptions(
+                () -> (Schema<T>) getStaticMethod(
+                        "org.apache.pulsar.client.impl.schema.ProtobufNativeSchema", "of", SchemaDefinition.class)
+                        .invoke(null, schemaDefinition));
+    }
+
     public static <T> Schema<T> newJSONSchema(SchemaDefinition schemaDefinition) {
         return catchExceptions(
                 () -> (Schema<T>) getStaticMethod(
@@ -326,10 +334,22 @@ public static Schema<?> getSchema(SchemaInfo schemaInfo) {
     }
 
     public static GenericSchema<GenericRecord> getGenericSchema(SchemaInfo schemaInfo) {
-        return catchExceptions(
-            () -> (GenericSchema) getStaticMethod(
-                "org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl",
-                "of", SchemaInfo.class).invoke(null, schemaInfo));
+        switch (schemaInfo.getType()) {
+            case PROTOBUF_NATIVE:
+                return (GenericSchema) ReflectionUtils.catchExceptions(() -> {
+                    return (GenericSchema) ReflectionUtils.getStaticMethod(
+                            "org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema",
+                            "of", new Class[]{SchemaInfo.class}).invoke((Object) null, schemaInfo);
+                });
+            default:
+                return (GenericSchema) ReflectionUtils.catchExceptions(() -> {
+                    return (GenericSchema) ReflectionUtils.getStaticMethod(
+                            "org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl",
+                            "of", new Class[]{SchemaInfo.class}).invoke((Object) null, schemaInfo);
+
+                });
+        }
+
     }
 
     public static RecordSchemaBuilder newRecordSchemaBuilder(String name) {
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java
index acbe34e58d431..e4311efe1fe67 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java
@@ -69,6 +69,7 @@ public String getSchemaDefinition() {
             case AVRO:
             case JSON:
             case PROTOBUF:
+            case PROTOBUF_NATIVE:
                 return new String(schema, UTF_8);
             case KEY_VALUE:
                 KeyValue<SchemaInfo, SchemaInfo> schemaInfoKeyValue =
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaType.java b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
index 3b16e73a3715a..24364013f0c13 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
@@ -134,6 +134,11 @@ public enum SchemaType {
      */
     LOCAL_DATE_TIME(19),
 
+    /**
+     * Protobuf native schema base on Descriptor.
+     */
+    PROTOBUF_NATIVE(20),
+
     //
     // Schemas that don't have schema info. the value should be negative.
     //
@@ -191,6 +196,7 @@ public static SchemaType valueOf(int value) {
           case 17: return LOCAL_DATE;
           case 18: return LOCAL_TIME;
           case 19: return LOCAL_DATE_TIME;
+          case 20: return PROTOBUF_NATIVE;
           case -1: return BYTES;
           case -2: return AUTO;
           case -3: return AUTO_CONSUME;
@@ -239,6 +245,7 @@ public static boolean isStructType(SchemaType type) {
             case AVRO:
             case JSON:
             case PROTOBUF:
+            case PROTOBUF_NATIVE:
                 return true;
             default:
                 return false;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java
index fb3ac59c31dbe..442ede177b37d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java
@@ -18,21 +18,22 @@
  */
 package org.apache.pulsar.client.impl.schema;
 
-import static com.google.common.base.Preconditions.checkState;
-
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SchemaSerializationException;
 import org.apache.pulsar.client.api.schema.GenericRecord;
 import org.apache.pulsar.client.api.schema.GenericSchema;
 import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
-import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
 import org.apache.pulsar.common.schema.KeyValue;
 import org.apache.pulsar.common.schema.SchemaInfo;
-import org.apache.pulsar.common.schema.SchemaType;
 
 import java.util.concurrent.ExecutionException;
 
+import static com.google.common.base.Preconditions.checkState;
+
 /**
  * Auto detect schema.
  */
@@ -147,13 +148,20 @@ public Schema<GenericRecord> clone() {
     }
 
     private GenericSchema generateSchema(SchemaInfo schemaInfo) {
-        if (schemaInfo.getType() != SchemaType.AVRO
-            && schemaInfo.getType() != SchemaType.JSON) {
-            throw new RuntimeException("Currently auto consume only works for topics with avro or json schemas");
-        }
         // when using `AutoConsumeSchema`, we use the schema associated with the messages as schema reader
         // to decode the messages.
-        return GenericSchemaImpl.of(schemaInfo, false /*useProvidedSchemaAsReaderSchema*/);
+        final boolean useProvidedSchemaAsReaderSchema = false;
+        switch (schemaInfo.getType()) {
+            case JSON:
+                return GenericJsonSchema.of(schemaInfo,useProvidedSchemaAsReaderSchema);
+            case AVRO:
+                return GenericAvroSchema.of(schemaInfo,useProvidedSchemaAsReaderSchema);
+            case PROTOBUF_NATIVE:
+                return GenericProtobufNativeSchema.of(schemaInfo, useProvidedSchemaAsReaderSchema);
+            default:
+                throw new IllegalArgumentException("Currently auto consume works for type '"
+                        + schemaInfo.getType() + "' is not supported yet");
+        }
     }
 
     public static Schema<?> getSchema(SchemaInfo schemaInfo) {
@@ -191,8 +199,11 @@ public static Schema<?> getSchema(SchemaInfo schemaInfo) {
             case LOCAL_DATE_TIME:
                 return LocalDateTimeSchema.of();
             case JSON:
+                return GenericJsonSchema.of(schemaInfo);
             case AVRO:
-                return GenericSchemaImpl.of(schemaInfo);
+                return GenericAvroSchema.of(schemaInfo);
+            case PROTOBUF_NATIVE:
+                return GenericProtobufNativeSchema.of(schemaInfo);
             case KEY_VALUE:
                 KeyValue<SchemaInfo, SchemaInfo> kvSchemaInfo =
                     KeyValueSchemaInfo.decodeKeyValueSchemaInfo(schemaInfo);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java
new file mode 100644
index 0000000000000..2a46155797db9
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java
@@ -0,0 +1,146 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.GeneratedMessageV3;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.api.schema.SchemaReader;
+import org.apache.pulsar.client.impl.schema.reader.ProtobufNativeReader;
+import org.apache.pulsar.client.impl.schema.writer.ProtobufNativeWriter;
+import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaType;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+/**
+ * A schema implementation to deal with protobuf generated messages.
+ */
+public class ProtobufNativeSchema<T extends GeneratedMessageV3> extends AbstractStructSchema<T> {
+
+    public static final String PARSING_INFO_PROPERTY = "__PARSING_INFO__";
+
+    @Getter
+    @AllArgsConstructor
+    public static class ProtoBufParsingInfo {
+        private final int number;
+        private final String name;
+        private final String type;
+        private final String label;
+        // For future nested fields
+        private final Map<String, Object> definition;
+    }
+
+    private static <T> Descriptors.Descriptor createProtobufNativeSchema(Class<T> pojo) {
+        try {
+            Method method = pojo.getMethod("getDescriptor");
+            Descriptors.Descriptor descriptor = (Descriptors.Descriptor) method.invoke(null);
+            return descriptor;
+        } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private ProtobufNativeSchema(SchemaInfo schemaInfo, T protoMessageInstance) {
+        super(schemaInfo);
+        setReader(new ProtobufNativeReader<>(protoMessageInstance));
+        setWriter(new ProtobufNativeWriter<>());
+        // update properties with protobuf related properties
+        Map<String, String> allProperties = new HashMap<>();
+        allProperties.putAll(schemaInfo.getProperties());
+        // set protobuf parsing info
+        allProperties.put(PARSING_INFO_PROPERTY, getParsingInfo(protoMessageInstance));
+        schemaInfo.setProperties(allProperties);
+    }
+
+    private String getParsingInfo(T protoMessageInstance) {
+        List<ProtoBufParsingInfo> protoBufParsingInfos = new LinkedList<>();
+        protoMessageInstance.getDescriptorForType().getFields().forEach(new Consumer<Descriptors.FieldDescriptor>() {
+            @Override
+            public void accept(Descriptors.FieldDescriptor fieldDescriptor) {
+                protoBufParsingInfos.add(new ProtoBufParsingInfo(fieldDescriptor.getNumber(),
+                        fieldDescriptor.getName(), fieldDescriptor.getType().name(),
+                        fieldDescriptor.toProto().getLabel().name(), null));
+            }
+        });
+
+        try {
+            return new ObjectMapper().writeValueAsString(protoBufParsingInfos);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public Descriptors.Descriptor getProtobufNativeSchema() {
+        return ProtobufNativeSchemaUtils.deserialize(this.schemaInfo.getSchema());
+    }
+
+    @Override
+    protected SchemaReader<T> loadReader(BytesSchemaVersion schemaVersion) {
+        throw new RuntimeException("ProtobufNativeSchema don't support schema versioning");
+    }
+
+    public static <T extends GeneratedMessageV3> ProtobufNativeSchema<T> of(Class<T> pojo) {
+        return of(pojo, new HashMap<>());
+    }
+
+    public static <T> ProtobufNativeSchema ofGenericClass(Class<T> pojo, Map<String, String> properties) {
+        SchemaDefinition<T> schemaDefinition = SchemaDefinition.<T>builder().withPojo(pojo).withProperties(properties).build();
+        return ProtobufNativeSchema.of(schemaDefinition);
+    }
+
+    public static <T> ProtobufNativeSchema of(SchemaDefinition<T> schemaDefinition) {
+        Class<T> pojo = schemaDefinition.getPojo();
+
+        if (!GeneratedMessageV3.class.isAssignableFrom(pojo)) {
+            throw new IllegalArgumentException(GeneratedMessageV3.class.getName()
+                    + " is not assignable from " + pojo.getName());
+        }
+        Descriptors.Descriptor descriptor = createProtobufNativeSchema(schemaDefinition.getPojo());
+
+        SchemaInfo schemaInfo = SchemaInfo.builder()
+                .schema(ProtobufNativeSchemaUtils.serialize(descriptor))
+                .type(SchemaType.PROTOBUF_NATIVE)
+                .name("")
+                .properties(schemaDefinition.getProperties())
+                .build();
+        try {
+            return new ProtobufNativeSchema(schemaInfo,
+                    (GeneratedMessageV3) pojo.getMethod("getDefaultInstance").invoke(null));
+        } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+    public static <T extends GeneratedMessageV3> ProtobufNativeSchema<T> of(
+            Class pojo, Map<String, String> properties) {
+        return ofGenericClass(pojo, properties);
+    }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
new file mode 100644
index 0000000000000..7ccef6b07ab9e
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
@@ -0,0 +1,139 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.common.protocol.schema.ProtobufNativeSchemaData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static com.google.protobuf.DescriptorProtos.FileDescriptorProto;
+import static com.google.protobuf.DescriptorProtos.FileDescriptorSet;
+import static com.google.protobuf.Descriptors.*;
+
+public class ProtobufNativeSchemaUtils {
+
+    public static byte[] serialize(Descriptor descriptor) {
+        byte[] schemaDataBytes;
+        try {
+            Map<String, FileDescriptorProto> fileDescriptorProtoCache = new HashMap<>();
+            //recursively cache all FileDescriptorProto
+            serializeFileDescriptor(descriptor.getFile(), fileDescriptorProtoCache);
+
+            //extract root message path
+            String rootMessageTypeName = descriptor.getFullName();
+            String rootFileDescriptorName = descriptor.getFile().getFullName();
+            //build FileDescriptorSet , this is equal to < protoc --include_imports --descriptor_set_out >
+            byte[] fileDescriptorSet = FileDescriptorSet.newBuilder().addAllFile(fileDescriptorProtoCache.values()).build().toByteArray();
+
+            //serialize to bytes
+            ProtobufNativeSchemaData schemaData = ProtobufNativeSchemaData.builder().fileDescriptorSet(fileDescriptorSet)
+                    .rootFileDescriptorName(rootFileDescriptorName).rootMessageTypeName(rootMessageTypeName).build();
+            schemaDataBytes = new ObjectMapper().writeValueAsBytes(schemaData);
+            logger.debug("descriptor : {} serialized to : {} ", descriptor.getFullName(), schemaDataBytes);
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw new SchemaSerializationException(e);
+        }
+        return schemaDataBytes;
+    }
+
+    private static void serializeFileDescriptor(FileDescriptor fileDescriptor, Map<String, FileDescriptorProto> fileDescriptorCache) {
+        fileDescriptor.getDependencies().forEach(dependency -> {
+                    if (!fileDescriptorCache.containsKey(dependency.getFullName())) {
+                        serializeFileDescriptor(dependency, fileDescriptorCache);
+                    }
+                }
+        );
+        String[] unResolvedFileDescriptNames = fileDescriptor.getDependencies().stream().
+                filter(item -> !fileDescriptorCache.containsKey(item.getFullName())).map(FileDescriptor::getFullName).toArray(String[]::new);
+        if (unResolvedFileDescriptNames.length == 0) {
+            fileDescriptorCache.put(fileDescriptor.getFullName(), fileDescriptor.toProto());
+        } else {
+            throw new SchemaSerializationException(fileDescriptor.getFullName() + " can't resolve dependency :" + unResolvedFileDescriptNames);
+        }
+    }
+
+    public static Descriptor deserialize(byte[] schemaDataBytes) {
+        Descriptor descriptor;
+        try {
+            ProtobufNativeSchemaData schemaData = new ObjectMapper().readValue(schemaDataBytes, ProtobufNativeSchemaData.class);
+
+            Map<String, FileDescriptorProto> fileDescriptorProtoCache = new HashMap<>();
+            Map<String, FileDescriptor> fileDescriptorCache = new HashMap<>();
+            FileDescriptorSet fileDescriptorSet = FileDescriptorSet.parseFrom(schemaData.getFileDescriptorSet());
+            fileDescriptorSet.getFileList().forEach(fileDescriptorProto -> {
+                fileDescriptorProtoCache.put(fileDescriptorProto.getName(), fileDescriptorProto);
+            });
+            FileDescriptorProto rootFileDescriptorProto = fileDescriptorProtoCache.get(schemaData.getRootFileDescriptorName());
+
+            //recursively build FileDescriptor
+            deserializeFileDescriptor(rootFileDescriptorProto, fileDescriptorCache, fileDescriptorProtoCache);
+            //extract root fileDescriptor
+            FileDescriptor fileDescriptor = fileDescriptorCache.get(schemaData.getRootFileDescriptorName());
+            //trim package
+            String[] paths = StringUtils.removeFirst(schemaData.getRootMessageTypeName(), fileDescriptor.getPackage()).replaceFirst(".", "").split("\\.");
+            //extract root message
+            descriptor = fileDescriptor.findMessageTypeByName(paths[0]);
+            //extract nested message
+            for (int i = 1; i < paths.length; i++) {
+                descriptor = descriptor.findNestedTypeByName(paths[i]);
+            }
+            logger.debug("deserialize : {} serialized to descriptor: {} ", schemaDataBytes, descriptor.getFullName());
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw new SchemaSerializationException(e);
+        }
+
+        return descriptor;
+    }
+
+    private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptorProto, Map<String, FileDescriptor> fileDescriptorCache, Map<String, FileDescriptorProto> fileDescriptorProtoCache) {
+        fileDescriptorProto.getDependencyList().forEach(dependencyFileDescriptorName -> {
+            if (!fileDescriptorCache.containsKey(dependencyFileDescriptorName)) {
+                FileDescriptorProto dependencyFileDescriptor = fileDescriptorProtoCache.get(dependencyFileDescriptorName);
+                deserializeFileDescriptor(dependencyFileDescriptor, fileDescriptorCache, fileDescriptorProtoCache);
+            }
+        });
+
+        FileDescriptor[] dependencyFileDescriptors = fileDescriptorProto.getDependencyList().stream().map(dependency -> {
+            if (fileDescriptorCache.containsKey(dependency)) {
+                return fileDescriptorCache.get(dependency);
+            } else {
+                throw new SchemaSerializationException(fileDescriptorProto.getName() + "can't resolve  dependency '" + dependency + "'");
+            }
+        }).toArray(FileDescriptor[]::n ew);
+
+        try {
+            FileDescriptor fileDescriptor = FileDescriptor.buildFrom(fileDescriptorProto, dependencyFileDescriptors);
+            fileDescriptorCache.put(fileDescriptor.getFullName(), fileDescriptor);
+        } catch (DescriptorValidationException e) {
+            e.printStackTrace();
+            throw new SchemaSerializationException(e);
+        }
+    }
+
+    private static final Logger logger = LoggerFactory.getLogger(ProtobufNativeSchemaUtils.class);
+
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReader.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReader.java
new file mode 100644
index 0000000000000..eccc48247cdd4
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReader.java
@@ -0,0 +1,83 @@
+/**
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.client.api.schema.Field;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.SchemaReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class GenericProtobufNativeReader implements SchemaReader<GenericRecord> {
+
+    private Descriptors.Descriptor descriptor;
+    private byte[] schemaVersion;
+    private List<Field> fields;
+
+    public GenericProtobufNativeReader(Descriptors.Descriptor descriptor) {
+        this(descriptor, null);
+    }
+
+    public GenericProtobufNativeReader(Descriptors.Descriptor descriptor, byte[] schemaVersion) {
+        try {
+            this.schemaVersion = schemaVersion;
+            this.descriptor = descriptor;
+            this.fields = descriptor.getFields()
+                    .stream()
+                    .map(f -> new Field(f.getName(), f.getIndex()))
+                    .collect(Collectors.toList());
+        } catch (Exception e) {
+            log.error("GenericProtobufNativeReader init error", e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public GenericProtobufNativeRecord read(byte[] bytes, int offset, int length) {
+        try {
+            if (!(bytes.length == length && offset == 0)) { //skip unnecessary bytes copy
+                bytes = Arrays.copyOfRange(bytes, offset, offset + length);
+            }
+            return new GenericProtobufNativeRecord(schemaVersion, descriptor, fields, DynamicMessage.parseFrom(descriptor, bytes));
+        } catch (InvalidProtocolBufferException e) {
+            throw new SchemaSerializationException(e);
+        }
+    }
+
+    @Override
+    public GenericProtobufNativeRecord read(InputStream inputStream) {
+        try {
+            return new GenericProtobufNativeRecord(schemaVersion, descriptor, fields, DynamicMessage.parseFrom(descriptor, inputStream));
+        } catch (IOException e) {
+            throw new SchemaSerializationException(e);
+        }
+    }
+
+    private static final Logger log = LoggerFactory.getLogger(GenericProtobufNativeReader.class);
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeRecord.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeRecord.java
new file mode 100644
index 0000000000000..51b4cb063da79
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeRecord.java
@@ -0,0 +1,46 @@
+/**
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
+import org.apache.pulsar.client.api.schema.Field;
+
+import java.util.List;
+
+public class GenericProtobufNativeRecord<T extends DynamicMessage> extends VersionedGenericRecord {
+
+    private DynamicMessage record;
+    private Descriptors.Descriptor msgDesc;
+
+    protected GenericProtobufNativeRecord(byte[] schemaVersion, Descriptors.Descriptor msgDesc, List<Field> fields, DynamicMessage record) {
+        super(schemaVersion, fields);
+        this.msgDesc = msgDesc;
+        this.record = record;
+    }
+
+    @Override
+    public Object getField(String fieldName) {
+        return record.getField(msgDesc.findFieldByName(fieldName));
+    }
+
+    public DynamicMessage getProtobufRecord() {
+        return record;
+    }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java
new file mode 100644
index 0000000000000..767350c0e4725
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java
@@ -0,0 +1,104 @@
+/**
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import com.google.protobuf.Descriptors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.schema.*;
+import org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaUtils;
+import org.apache.pulsar.client.impl.schema.SchemaUtils;
+import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class GenericProtobufNativeSchema extends AbstractGenericSchema {
+
+    Descriptors.Descriptor descriptor;
+
+    public GenericProtobufNativeSchema(SchemaInfo schemaInfo) {
+        this(schemaInfo, true);
+    }
+
+    public GenericProtobufNativeSchema(SchemaInfo schemaInfo,
+                                       boolean useProvidedSchemaAsReaderSchema) {
+        super(schemaInfo, useProvidedSchemaAsReaderSchema);
+        this.descriptor = parseProtobufSchema(schemaInfo);
+        this.fields = descriptor.getFields()
+                .stream()
+                .map(f -> new Field(f.getName(), f.getIndex()))
+                .collect(Collectors.toList());
+        setReader(new GenericProtobufNativeReader(descriptor));
+        setWriter(new GenericProtobufNativeWriter());
+    }
+
+    @Override
+    public List<Field> getFields() {
+        return fields;
+    }
+
+    @Override
+    public GenericRecordBuilder newRecordBuilder() {
+        return new ProtobufNativeRecordBuilderImpl(this);
+    }
+
+    @Override
+    protected SchemaReader<GenericRecord> loadReader(BytesSchemaVersion schemaVersion) {
+        SchemaInfo schemaInfo = getSchemaInfoByVersion(schemaVersion.get());
+        if (schemaInfo != null) {
+            log.info("Load schema reader for version({}), schema is : {}",
+                    SchemaUtils.getStringSchemaVersion(schemaVersion.get()),
+                    schemaInfo);
+            Descriptors.Descriptor recordDescriptor = parseProtobufSchema(schemaInfo);
+            Descriptors.Descriptor readerSchemaDescriptor = useProvidedSchemaAsReaderSchema ? descriptor : recordDescriptor;
+            return new GenericProtobufNativeReader(
+                    readerSchemaDescriptor,
+                    schemaVersion.get());
+        } else {
+            log.warn("No schema found for version({}), use latest schema : {}",
+                    SchemaUtils.getStringSchemaVersion(schemaVersion.get()),
+                    this.schemaInfo);
+            return reader;
+        }
+    }
+
+    protected static Descriptors.Descriptor parseProtobufSchema(SchemaInfo schemaInfo) {
+        return ProtobufNativeSchemaUtils.deserialize(schemaInfo.getSchema());
+    }
+
+    public static GenericSchema of(SchemaInfo schemaInfo) {
+        return new GenericProtobufNativeSchema(schemaInfo, true);
+    }
+
+    public static GenericSchema of(SchemaInfo schemaInfo, boolean useProvidedSchemaAsReaderSchema) {
+        return new GenericProtobufNativeSchema(schemaInfo, useProvidedSchemaAsReaderSchema);
+    }
+
+    public Descriptors.Descriptor getProtobufNativeSchema() {
+        return descriptor;
+    }
+
+    @Override
+    public boolean supportSchemaVersioning() {
+        return true;
+    }
+
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeWriter.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeWriter.java
new file mode 100644
index 0000000000000..e81b692eaf430
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeWriter.java
@@ -0,0 +1,29 @@
+/**
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.SchemaWriter;
+
+public class GenericProtobufNativeWriter implements SchemaWriter<GenericRecord> {
+    @Override
+    public byte[] write(GenericRecord message) {
+        return ((GenericProtobufNativeRecord) message).getProtobufRecord().toByteArray();
+    }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/ProtobufNativeRecordBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/ProtobufNativeRecordBuilderImpl.java
new file mode 100644
index 0000000000000..e9f333ffeb662
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/ProtobufNativeRecordBuilderImpl.java
@@ -0,0 +1,72 @@
+/**
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
+import org.apache.pulsar.client.api.schema.Field;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericRecordBuilder;
+
+public class ProtobufNativeRecordBuilderImpl implements GenericRecordBuilder {
+
+    private GenericProtobufNativeSchema genericSchema;
+    private DynamicMessage.Builder builder;
+    private Descriptors.Descriptor msgDesc;
+
+    public ProtobufNativeRecordBuilderImpl(GenericProtobufNativeSchema genericSchema) {
+        this.genericSchema = genericSchema;
+        this.msgDesc = genericSchema.getProtobufNativeSchema();
+        builder = DynamicMessage.newBuilder(msgDesc);
+    }
+
+    @Override
+    public GenericRecordBuilder set(String fieldName, Object value) {
+        builder.setField(msgDesc.findFieldByName(fieldName), value);
+        return this;
+    }
+
+    @Override
+    public GenericRecordBuilder set(Field field, Object value) {
+        builder.setField(msgDesc.findFieldByName(field.getName()), value);
+        return this;
+    }
+
+    @Override
+    public GenericRecordBuilder clear(String fieldName) {
+        builder.clearField(msgDesc.findFieldByName(fieldName));
+        return this;
+    }
+
+    @Override
+    public GenericRecordBuilder clear(Field field) {
+        builder.clearField(msgDesc.findFieldByName(field.getName()));
+        return this;
+    }
+
+    @Override
+    public GenericRecord build() {
+        return new GenericProtobufNativeRecord(
+                null,
+                genericSchema.getProtobufNativeSchema(),
+                genericSchema.getFields(),
+                builder.build()
+        );
+    }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java
new file mode 100644
index 0000000000000..9701e2ffa1b13
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java
@@ -0,0 +1,27 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema.reader;
+
+public class ProtobufNativeReader<T extends com.google.protobuf.GeneratedMessageV3> extends ProtobufReader<T> {
+
+    public ProtobufNativeReader(T protoMessageInstance) {
+        super(protoMessageInstance);
+    }
+
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java
new file mode 100644
index 0000000000000..82f241c9370bd
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java
@@ -0,0 +1,27 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema.writer;
+
+public class ProtobufNativeWriter<T extends com.google.protobuf.GeneratedMessageV3> extends ProtobufWriter<T> {
+
+    public ProtobufNativeWriter() {
+        super();
+    }
+
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java
index 9516c25217f11..c5d5f338f1b49 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java
@@ -451,6 +451,7 @@ public enum Type
       LocalDate(17, 17),
       LocalTime(18, 18),
       LocalDateTime(19, 19),
+      ProtobufNative(20, 20),
       ;
       
       public static final int None_VALUE = 0;
@@ -473,6 +474,7 @@ public enum Type
       public static final int LocalDate_VALUE = 17;
       public static final int LocalTime_VALUE = 18;
       public static final int LocalDateTime_VALUE = 19;
+      public static final int ProtobufNative_VALUE = 20;
       
       
       public final int getNumber() { return value; }
@@ -499,6 +501,7 @@ public static Type valueOf(int value) {
           case 17: return LocalDate;
           case 18: return LocalTime;
           case 19: return LocalDateTime;
+          case 20: return ProtobufNative;
           default: return null;
         }
       }
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/ProtobufNativeSchemaData.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/ProtobufNativeSchemaData.java
new file mode 100644
index 0000000000000..069b95c0ed0cf
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/ProtobufNativeSchemaData.java
@@ -0,0 +1,47 @@
+/**
+ * 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 org.apache.pulsar.common.protocol.schema;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * POJO class used for serialize to json-string for SchemaInfo.schema when SchemaType is SchemaType.PROTOBUF_NATIVE.
+ */
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class ProtobufNativeSchemaData {
+    /**
+     * protobuf v3 FileDescriptorSet bytes.
+     **/
+    private byte[] fileDescriptorSet;
+    /**
+     * protobuf v3 rootMessageTypeName.
+     **/
+    private String rootMessageTypeName;
+    /**
+     * protobuf v3 rootFileDescriptorName.
+     **/
+    private String rootFileDescriptorName;
+
+}
diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto
index e740b1120b4c3..738416e0a1772 100644
--- a/pulsar-common/src/main/proto/PulsarApi.proto
+++ b/pulsar-common/src/main/proto/PulsarApi.proto
@@ -44,6 +44,7 @@ message Schema {
         LocalDate = 17;
         LocalTime = 18;
         LocalDateTime = 19;
+        ProtobufNative = 20;
     }
 
     required string name = 1;

From 4bc698b76317dd148e776c5c9df591f40298cc41 Mon Sep 17 00:00:00 2001
From: wangguowei <wangguowei@zhihu.com>
Date: Tue, 27 Oct 2020 11:02:22 +0800
Subject: [PATCH 02/10] commit unit test

---
 ...rotobufNativeSchemaCompatibilityCheck.java |   2 +-
 ...bufNativeSchemaCompatibilityCheckTest.java |  50 +++++++++
 .../schema/ProtobufNativeSchemaUtils.java     |   2 +-
 .../impl/schema/ProtobufNativeSchemaTest.java | 103 ++++++++++++++++++
 .../schema/ProtobufNativeSchemaUtilsTest.java |  44 ++++++++
 .../impl/schema/ProtobufSchemaTest.java       |  36 +++---
 .../generic/AbstractGenericSchemaTest.java    |  98 +++++++++++++++++
 .../GenericProtobufNativeReaderTest.java      |  71 ++++++++++++
 .../GenericProtobufNativeSchemaTest.java      |  66 +++++++++++
 .../src/test/proto/ExternalTest.proto         |  28 +++++
 pulsar-client/src/test/proto/Test.proto       |   8 ++
 .../io/influxdb/v2/InfluxDBSinkTest.java      |   4 +-
 12 files changed, 492 insertions(+), 20 deletions(-)
 create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java
 create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java
 create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java
 create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
 create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
 create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java
 create mode 100644 pulsar-client/src/test/proto/ExternalTest.proto

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
index e8d712bbc5253..4550a076284fa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
@@ -39,7 +39,7 @@ public SchemaType getSchemaType() {
     @Override
     public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy) throws IncompatibleSchemaException {
         Descriptor fromDescriptor = ProtobufNativeSchemaUtils.deserialize(from.getData());
-        Descriptor toDescriptor = ProtobufNativeSchemaUtils.deserialize(from.getData());
+        Descriptor toDescriptor = ProtobufNativeSchemaUtils.deserialize(to.getData());
         switch (strategy) {
             case BACKWARD_TRANSITIVE:
             case BACKWARD:
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java
new file mode 100644
index 0000000000000..75f6b6dd76e04
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java
@@ -0,0 +1,50 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaUtils;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static com.google.protobuf.Descriptors.Descriptor;
+
+public class ProtobufNativeSchemaCompatibilityCheckTest {
+
+    private static final SchemaData schemaData1 = getSchemaData(org.apache.pulsar.client.api.schema.proto.Test.TestMessage.getDescriptor());
+
+    private static final SchemaData schemaData2 = getSchemaData(org.apache.pulsar.client.api.schema.proto.Test.SubMessage.getDescriptor());
+
+    /**
+     * make sure protobuf root message isn't allow change
+     */
+    @Test
+    public void testRootMessageChange() {
+        ProtobufNativeSchemaCompatibilityCheck compatibilityCheck = new ProtobufNativeSchemaCompatibilityCheck();
+        Assert.assertFalse(compatibilityCheck.isCompatible(schemaData2, schemaData1,
+                SchemaCompatibilityStrategy.FULL),
+                "Protobuf root message isn't allow change");
+    }
+
+    private static SchemaData getSchemaData(Descriptor descriptor) {
+        return SchemaData.builder().data(ProtobufNativeSchemaUtils.serialize(descriptor)).type(SchemaType.PROTOBUF_NATIVE).build();
+    }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
index 7ccef6b07ab9e..3c8e51111f2d4 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
@@ -123,7 +123,7 @@ private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptor
             } else {
                 throw new SchemaSerializationException(fileDescriptorProto.getName() + "can't resolve  dependency '" + dependency + "'");
             }
-        }).toArray(FileDescriptor[]::n ew);
+        }).toArray(FileDescriptor[]::new);
 
         try {
             FileDescriptor fileDescriptor = FileDescriptor.buildFrom(fileDescriptorProto, dependencyFileDescriptors);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java
new file mode 100644
index 0000000000000..f1b0b65d18992
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java
@@ -0,0 +1,103 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+
+@Slf4j
+public class ProtobufNativeSchemaTest {
+
+
+    private static final String EXPECTED_SCHEMA_JSON = "{\"fileDescriptorSet\":\"CtMDCgpUZXN0LnByb3RvEgVwcm90bxoSRXh0ZXJuYWxUZXN0LnByb3RvImUKClN1Yk1lc3NhZ2U" +
+            "SCwoDZm9vGAEgASgJEgsKA2JhchgCIAEoARo9Cg1OZXN0ZWRNZXNzYWdlEgsKA3VybBgBIAEoCRINCgV0aXRsZRgCIAEoCRIQCghzbmlwcGV0cxgDIAMoCSLlAQoLV" +
+            "GVzdE1lc3NhZ2USEwoLc3RyaW5nRmllbGQYASABKAkSEwoLZG91YmxlRmllbGQYAiABKAESEAoIaW50RmllbGQYBiABKAUSIQoIdGVzdEVudW0YBCABKA4yDy5wcm90by5U" +
+            "ZXN0RW51bRImCgtuZXN0ZWRGaWVsZBgFIAEoCzIRLnByb3RvLlN1Yk1lc3NhZ2USFQoNcmVwZWF0ZWRGaWVsZBgKIAMoCRI4Cg9leHRlcm5hbE1lc3NhZ2UYCyABKAsyHy5wcm90by" +
+            "5leHRlcm5hbC5FeHRlcm5hbE1lc3NhZ2UqJAoIVGVzdEVudW0SCgoGU0hBUkVEEAASDAoIRkFJTE9WRVIQAUItCiVvcmcuYXBhY2hlLnB1bHNhci5jbGllbnQuc2NoZW1hLnByb3" +
+            "RvQgRUZXN0YgZwcm90bzMKoAEKEkV4dGVybmFsVGVzdC5wcm90bxIOcHJvdG8uZXh0ZXJuYWwiOwoPRXh0ZXJuYWxNZXNzYWdlEhMKC3N0cmluZ0ZpZWxkGAEgA" +
+            "SgJEhMKC2RvdWJsZUZpZWxkGAIgASgBQjUKJW9yZy5hcGFjaGUucHVsc2FyLmNsaWVudC5zY2hlbWEucHJvdG9CDEV4dGVybmFsVGVzdGIGcHJvdG8z\"," +
+            "\"rootMessageTypeName\":\"proto.TestMessage\",\"rootFileDescriptorName\":\"Test.proto\"}";
+
+    @Test
+    public void testEncodeAndDecode() {
+        final String stringFieldValue = "StringFieldValue";
+        org.apache.pulsar.client.schema.proto.Test.TestMessage testMessage = org.apache.pulsar.client.schema.proto.Test.TestMessage.newBuilder().setStringField(stringFieldValue).build();
+        ProtobufNativeSchema<org.apache.pulsar.client.schema.proto.Test.TestMessage> protobufSchema = ProtobufNativeSchema.of(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
+
+        byte[] bytes = protobufSchema.encode(testMessage);
+        org.apache.pulsar.client.schema.proto.Test.TestMessage message = protobufSchema.decode(bytes);
+
+        Assert.assertEquals(message.getStringField(), stringFieldValue);
+    }
+
+    @Test
+    public void testSchema() {
+        ProtobufNativeSchema<org.apache.pulsar.client.schema.proto.Test.TestMessage> protobufSchema
+                = ProtobufNativeSchema.of(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
+
+        Assert.assertEquals(protobufSchema.getSchemaInfo().getType(), SchemaType.PROTOBUF_NATIVE);
+
+        Assert.assertNotNull(ProtobufNativeSchemaUtils.deserialize(protobufSchema.getSchemaInfo().getSchema()));
+        Assert.assertEquals(new String(protobufSchema.getSchemaInfo().getSchema(), StandardCharsets.UTF_8), EXPECTED_SCHEMA_JSON);
+    }
+
+    @Test
+    public void testGenericOf() {
+        try {
+            ProtobufNativeSchema<org.apache.pulsar.client.schema.proto.Test.TestMessage> protobufNativeSchema
+                    = ProtobufNativeSchema.ofGenericClass(org.apache.pulsar.client.schema.proto.Test.TestMessage.class,
+                    new HashMap<>());
+        } catch (Exception e) {
+            Assert.fail("Should not construct a ProtobufShema over a non-protobuf-generated class");
+        }
+
+        try {
+            ProtobufSchema<org.apache.pulsar.client.schema.proto.Test.TestMessage> protobufSchema
+                    = ProtobufSchema.ofGenericClass(String.class,
+                    Collections.emptyMap());
+            Assert.fail("Should not construct a ProtobufNativeShema over a non-protobuf-generated class");
+        } catch (Exception e) {
+
+        }
+    }
+
+
+    @Test
+    public void testDecodeByteBuf() {
+        ProtobufNativeSchema<org.apache.pulsar.client.schema.proto.Test.TestMessage> protobufSchema
+                = ProtobufNativeSchema.of(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
+        org.apache.pulsar.client.schema.proto.Test.TestMessage testMessage =
+                org.apache.pulsar.client.schema.proto.Test.TestMessage.newBuilder().build();
+        byte[] bytes = protobufSchema.encode(org.apache.pulsar.client.schema.proto.Test.TestMessage.newBuilder().build());
+        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(bytes.length);
+        byteBuf.writeBytes(bytes);
+
+        Assert.assertEquals(testMessage, protobufSchema.decode(byteBuf));
+
+    }
+
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java
new file mode 100644
index 0000000000000..29ab73ac5c008
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtilsTest.java
@@ -0,0 +1,44 @@
+/**
+ * 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 org.apache.pulsar.client.impl.schema;
+
+import com.google.protobuf.Descriptors;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class ProtobufNativeSchemaUtilsTest {
+
+    @Test
+    public static void testSerialize() throws Exception {
+        byte[] data =  ProtobufNativeSchemaUtils.serialize(org.apache.pulsar.client.schema.proto.Test.TestMessage.getDescriptor());
+        Descriptors.Descriptor descriptor =  ProtobufNativeSchemaUtils.deserialize(data);
+        Assert.assertNotNull(descriptor);
+        Assert.assertNotNull(descriptor.findFieldByName("nestedField").getMessageType());
+        Assert.assertNotNull(descriptor.findFieldByName("externalMessage").getMessageType());
+    }
+
+    @Test
+    public static void testNestedMessage() throws Exception {
+        byte[] data =  ProtobufNativeSchemaUtils.serialize(org.apache.pulsar.client.schema.proto.Test.SubMessage.NestedMessage.getDescriptor());
+        Descriptors.Descriptor descriptor =  ProtobufNativeSchemaUtils.deserialize(data);
+        Assert.assertNotNull(descriptor);
+
+    }
+
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufSchemaTest.java
index 4c353565202af..cb8138dd86646 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufSchemaTest.java
@@ -40,23 +40,27 @@ public class ProtobufSchemaTest {
     private static final String EXPECTED_SCHEMA_JSON = "{\"type\":\"record\",\"name\":\"TestMessage\"," +
             "\"namespace\":\"org.apache.pulsar.client.schema.proto.Test\",\"fields\":[{\"name\":\"stringField\"," +
             "\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"},\"default\":\"\"}," +
-            "{\"name\":\"doubleField\",\"type\":\"double\",\"default\":0},{\"name\":\"intField\",\"type\":\"int\"," +
-            "\"default\":0},{\"name\":\"testEnum\",\"type\":{\"type\":\"enum\",\"name\":\"TestEnum\"," +
-            "\"symbols\":[\"SHARED\",\"FAILOVER\"]},\"default\":\"SHARED\"},{\"name\":\"nestedField\"," +
-            "\"type\":[\"null\",{\"type\":\"record\",\"name\":\"SubMessage\",\"fields\":[{\"name\":\"foo\"," +
-            "\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"},\"default\":\"\"},{\"name\":\"bar\"," +
-            "\"type\":\"double\",\"default\":0}]}],\"default\":null},{\"name\":\"repeatedField\"," +
-            "\"type\":{\"type\":\"array\",\"items\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}}]}";
-
-    private static final String EXPECTED_PARSING_INFO = "{\"__alwaysAllowNull\":\"true\",\"__jsr310ConversionEnabled\":\"false\",\"__PARSING_INFO__\":" +
-            "\"[{\\\"number\\\":1,\\\"name\\\":\\\"stringField\\\",\\\"type\\\":\\\"STRING\\\",\\\"label\\\":\\\"" +
-            "LABEL_OPTIONAL\\\",\\\"definition\\\":null},{\\\"number\\\":2,\\\"name\\\":\\\"doubleField\\\",\\\"type\\\"" +
-            ":\\\"DOUBLE\\\",\\\"label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null},{\\\"number\\\":6,\\\"name\\\"" +
-            ":\\\"intField\\\",\\\"type\\\":\\\"INT32\\\",\\\"label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null}," +
+            "{\"name\":\"doubleField\",\"type\":\"double\",\"default\":0},{\"name\":\"intField\"," +
+            "\"type\":\"int\",\"default\":0},{\"name\":\"testEnum\",\"type\":{\"type\":\"enum\",\"name\":\"TestEnum\"," +
+            "\"symbols\":[\"SHARED\",\"FAILOVER\"]},\"default\":\"SHARED\"},{\"name\":\"nestedField\",\"type\":[\"null\"," +
+            "{\"type\":\"record\",\"name\":\"SubMessage\",\"fields\":[{\"name\":\"foo\",\"type\":{\"type\":\"string\"," +
+            "\"avro.java.string\":\"String\"},\"default\":\"\"},{\"name\":\"bar\",\"type\":\"double\",\"default\":0}]}]," +
+            "\"default\":null},{\"name\":\"repeatedField\",\"type\":{\"type\":\"array\",\"items\":{\"type\":\"string\"," +
+            "\"avro.java.string\":\"String\"}}},{\"name\":\"externalMessage\",\"type\":[\"null\",{\"type\":\"record\"," +
+            "\"name\":\"ExternalMessage\",\"namespace\":\"org.apache.pulsar.client.schema.proto.ExternalTest\"," +
+            "\"fields\":[{\"name\":\"stringField\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}," +
+            "\"default\":\"\"},{\"name\":\"doubleField\",\"type\":\"double\",\"default\":0}]}],\"default\":null}]}";
+
+    private static final String EXPECTED_PARSING_INFO = "{\"__alwaysAllowNull\":\"true\",\"__jsr310ConversionEnabled\":\"false\"," +
+            "\"__PARSING_INFO__\":\"[{\\\"number\\\":1,\\\"name\\\":\\\"stringField\\\",\\\"type\\\":\\\"STRING\\\"," +
+            "\\\"label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null},{\\\"number\\\":2,\\\"name\\\":\\\"doubleField\\\"," +
+            "\\\"type\\\":\\\"DOUBLE\\\",\\\"label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null},{\\\"number\\\":6," +
+            "\\\"name\\\":\\\"intField\\\",\\\"type\\\":\\\"INT32\\\",\\\"label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null}," +
             "{\\\"number\\\":4,\\\"name\\\":\\\"testEnum\\\",\\\"type\\\":\\\"ENUM\\\",\\\"label\\\":\\\"LABEL_OPTIONAL\\\"," +
-            "\\\"definition\\\":null},{\\\"number\\\":5,\\\"name\\\":\\\"nestedField\\\",\\\"type\\\":\\\"MESSAGE\\\",\\\"" +
-            "label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null},{\\\"number\\\":10,\\\"name\\\":\\\"repeatedField\\\"," +
-            "\\\"type\\\":\\\"STRING\\\",\\\"label\\\":\\\"LABEL_REPEATED\\\",\\\"definition\\\":null}]\"}";
+            "\\\"definition\\\":null},{\\\"number\\\":5,\\\"name\\\":\\\"nestedField\\\",\\\"type\\\":\\\"MESSAGE\\\"," +
+            "\\\"label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null},{\\\"number\\\":10,\\\"name\\\":\\\"repeatedField\\\"," +
+            "\\\"type\\\":\\\"STRING\\\",\\\"label\\\":\\\"LABEL_REPEATED\\\",\\\"definition\\\":null},{\\\"number\\\":11," +
+            "\\\"name\\\":\\\"externalMessage\\\",\\\"type\\\":\\\"MESSAGE\\\",\\\"label\\\":\\\"LABEL_OPTIONAL\\\",\\\"definition\\\":null}]\"}";
 
     @Test
     public void testEncodeAndDecode() {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
new file mode 100644
index 0000000000000..f2e59b978fa9e
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
@@ -0,0 +1,98 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
+import org.testng.annotations.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.mockito.Mockito.*;
+import static org.testng.Assert.assertEquals;
+
+/**
+ * Unit testing AbstractGenericSchema for non-avroBasedGenericSchema
+ */
+@Slf4j
+public class AbstractGenericSchemaTest {
+
+    @Test
+    public void testGenericProtobufNativeSchema() {
+        Schema<org.apache.pulsar.client.schema.proto.Test.TestMessage> encodeSchema = Schema.PROTOBUFNATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
+        GenericSchema decodeSchema = GenericProtobufNativeSchema.of(encodeSchema.getSchemaInfo());
+
+        testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema);
+    }
+
+    @Test
+    public void testAutoProtobufNativeSchema() {
+        // configure the schema info provider
+        MultiVersionSchemaInfoProvider multiVersionSchemaInfoProvider = mock(MultiVersionSchemaInfoProvider.class);
+        GenericSchema genericProtobufNativeSchema = GenericProtobufNativeSchema.of(Schema.PROTOBUFNATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class).getSchemaInfo());
+        when(multiVersionSchemaInfoProvider.getSchemaByVersion(any(byte[].class)))
+                .thenReturn(CompletableFuture.completedFuture(genericProtobufNativeSchema.getSchemaInfo()));
+
+        // configure encode schema
+        Schema<org.apache.pulsar.client.schema.proto.Test.TestMessage> encodeSchema = Schema.PROTOBUFNATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
+        // configure decode schema
+        AutoConsumeSchema decodeSchema = new AutoConsumeSchema();
+        decodeSchema.configureSchemaInfo("test-topic", "topic", encodeSchema.getSchemaInfo());
+        decodeSchema.setSchemaInfoProvider(multiVersionSchemaInfoProvider);
+
+        testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema);
+    }
+
+    private void testEncodeAndDecodeGenericRecord(Schema<org.apache.pulsar.client.schema.proto.Test.TestMessage> encodeSchema,
+                                                  Schema<GenericRecord> decodeSchema) {
+        int numRecords = 10;
+        for (int i = 0; i < numRecords; i++) {
+            org.apache.pulsar.client.schema.proto.Test.TestMessage testMessage = newTestMessage(i);
+            byte[] data = encodeSchema.encode(testMessage);
+
+            log.info("Decoding : {}", new String(data, UTF_8));
+
+            GenericRecord record;
+            if (decodeSchema instanceof AutoConsumeSchema) {
+                record = decodeSchema.decode(data, new byte[0]);
+            } else {
+                record = decodeSchema.decode(data);
+            }
+            verifyTestMessageRecord(record, i);
+        }
+    }
+
+
+    private static org.apache.pulsar.client.schema.proto.Test.TestMessage newTestMessage(int i) {
+        return org.apache.pulsar.client.schema.proto.Test.TestMessage.newBuilder().setStringField("field-value-" + i)
+                .setIntField(i).build();
+    }
+
+    private static void verifyTestMessageRecord(GenericRecord record, int i) {
+        Object stringField = record.getField("stringField");
+        assertEquals("field-value-" + i, stringField);
+        Object intField = record.getField("intField");
+        assertEquals(+i, intField);
+    }
+
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
new file mode 100644
index 0000000000000..b51c3c62984fe
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
@@ -0,0 +1,71 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.impl.schema.ProtobufNativeSchema;
+import org.apache.pulsar.client.schema.proto.Test.TestMessage;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+@Slf4j
+public class GenericProtobufNativeReaderTest {
+
+    private TestMessage message;
+    private GenericRecord genericmessage;
+    private GenericProtobufNativeSchema genericProtobufNativeSchema;
+    private ProtobufNativeSchema clazzBasedProtobufNativeSchema;
+
+
+    @BeforeMethod
+    public void setup() {
+        clazzBasedProtobufNativeSchema = ProtobufNativeSchema.of(SchemaDefinition.<TestMessage>builder()
+                .withPojo(TestMessage.class).build());
+        genericProtobufNativeSchema = (GenericProtobufNativeSchema) GenericProtobufNativeSchema.of(clazzBasedProtobufNativeSchema.getSchemaInfo());
+
+    }
+
+    @Test
+    public void testGenericAvroReaderByClazzBasedWriterSchema() {
+        message = TestMessage.newBuilder().setStringField(STRING_FIELD_VLUE).setDoubleField(DOUBLE_FIELD_VLUE).build();
+        GenericProtobufNativeReader genericProtobufNativeReader = new GenericProtobufNativeReader(genericProtobufNativeSchema.getProtobufNativeSchema());
+        GenericRecord genericRecordByWriterSchema = genericProtobufNativeReader.read(message.toByteArray());
+        assertEquals(genericRecordByWriterSchema.getField("stringField"), STRING_FIELD_VLUE);
+        assertEquals(genericRecordByWriterSchema.getField("doubleField"), DOUBLE_FIELD_VLUE);
+    }
+
+    @Test
+    public void testClazzBasedReaderByGenericWriterSchema() {
+        genericmessage = genericProtobufNativeSchema.newRecordBuilder().set("stringField", STRING_FIELD_VLUE).set("doubleField", DOUBLE_FIELD_VLUE).build();
+        byte[] messageBytes = new GenericProtobufNativeWriter().write(genericmessage);
+        GenericProtobufNativeReader genericProtobufNativeReader = new GenericProtobufNativeReader(clazzBasedProtobufNativeSchema.getProtobufNativeSchema());
+        GenericRecord genericRecordByWriterSchema = genericProtobufNativeReader.read(messageBytes);
+        assertEquals(genericRecordByWriterSchema.getField("stringField"), STRING_FIELD_VLUE);
+        assertEquals(genericRecordByWriterSchema.getField("doubleField"), DOUBLE_FIELD_VLUE);
+
+    }
+
+    private final static String STRING_FIELD_VLUE = "stringFieldValue";
+    private final static double DOUBLE_FIELD_VLUE = 0.2D;
+
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java
new file mode 100644
index 0000000000000..ac903d15a6969
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java
@@ -0,0 +1,66 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.pulsar.client.impl.schema.generic;
+
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.impl.schema.ProtobufNativeSchema;
+import org.apache.pulsar.client.schema.proto.Test.TestMessage;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+public class GenericProtobufNativeSchemaTest {
+
+    private TestMessage message;
+    private GenericRecord genericmessage;
+    private GenericProtobufNativeSchema genericProtobufNativeSchema;
+    private ProtobufNativeSchema<TestMessage> clazzBasedProtobufNativeSchema;
+
+    @BeforeMethod
+    public void init() {
+        clazzBasedProtobufNativeSchema = ProtobufNativeSchema.of(SchemaDefinition.<TestMessage>builder()
+                .withPojo(TestMessage.class).build());
+        genericProtobufNativeSchema = (GenericProtobufNativeSchema) GenericProtobufNativeSchema.of(clazzBasedProtobufNativeSchema.getSchemaInfo());
+
+    }
+
+    @Test
+    public void testGenericReaderByClazzBasedWriterSchema() {
+        message = TestMessage.newBuilder().setStringField(STRING_FIELD_VLUE).setDoubleField(DOUBLE_FIELD_VLUE).build();
+        byte[] clazzBasedProtobufBytes = clazzBasedProtobufNativeSchema.encode(message);
+        GenericRecord genericRecord = genericProtobufNativeSchema.decode(clazzBasedProtobufBytes);
+        assertEquals(genericRecord.getField("stringField"), STRING_FIELD_VLUE);
+        assertEquals(genericRecord.getField("doubleField"), DOUBLE_FIELD_VLUE);
+    }
+
+    @Test
+    public void testClazzBasedReaderByClazzGenericWriterSchema() {
+        genericmessage = genericProtobufNativeSchema.newRecordBuilder().set("stringField", STRING_FIELD_VLUE).set("doubleField", DOUBLE_FIELD_VLUE).build();
+        byte[] messageBytes = genericProtobufNativeSchema.encode(genericmessage);
+        message = clazzBasedProtobufNativeSchema.decode(messageBytes);
+        assertEquals(message.getStringField(), STRING_FIELD_VLUE);
+        assertEquals(message.getDoubleField(), DOUBLE_FIELD_VLUE);
+    }
+
+    private final static String STRING_FIELD_VLUE = "stringFieldValue";
+    private final static double DOUBLE_FIELD_VLUE = 0.2D;
+
+}
diff --git a/pulsar-client/src/test/proto/ExternalTest.proto b/pulsar-client/src/test/proto/ExternalTest.proto
new file mode 100644
index 0000000000000..dac5d48038590
--- /dev/null
+++ b/pulsar-client/src/test/proto/ExternalTest.proto
@@ -0,0 +1,28 @@
+/**
+ * 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.
+ */
+syntax = "proto3";
+package proto.external;
+
+option java_package = "org.apache.pulsar.client.schema.proto";
+option java_outer_classname = "ExternalTest";
+
+message ExternalMessage {
+    string stringField = 1;
+    double doubleField = 2;
+}
\ No newline at end of file
diff --git a/pulsar-client/src/test/proto/Test.proto b/pulsar-client/src/test/proto/Test.proto
index 7d7b1b64ab2d7..00515e9404db0 100644
--- a/pulsar-client/src/test/proto/Test.proto
+++ b/pulsar-client/src/test/proto/Test.proto
@@ -19,6 +19,8 @@
 syntax = "proto3";
 package proto;
 
+import "ExternalTest.proto";
+
 option java_package = "org.apache.pulsar.client.schema.proto";
 option java_outer_classname = "Test";
 
@@ -30,6 +32,11 @@ enum TestEnum {
 message SubMessage {
     string foo = 1;
     double bar = 2;
+    message NestedMessage {
+        string url = 1;
+        string title = 2;
+        repeated string snippets = 3;
+    }
 }
 
 message TestMessage {
@@ -39,4 +46,5 @@ message TestMessage {
     TestEnum testEnum = 4;
     SubMessage nestedField = 5;
     repeated string repeatedField = 10;
+    proto.external.ExternalMessage externalMessage = 11;
 }
\ No newline at end of file
diff --git a/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java b/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java
index a2e77cde5267f..02919aa8aba41 100644
--- a/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java
+++ b/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java
@@ -31,7 +31,7 @@
 import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
 import org.apache.pulsar.client.impl.schema.AvroSchema;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
-import org.apache.pulsar.client.impl.schema.StructSchema;
+import org.apache.pulsar.client.impl.schema.AbstractStructSchema;
 import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
 import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
 import org.apache.pulsar.functions.api.Record;
@@ -144,7 +144,7 @@ public void testOpenWriteCloseJson() throws Exception {
         openWriteClose(jsonSchema);
     }
 
-    private void openWriteClose(StructSchema<Cpu> schema) throws Exception {
+    private void openWriteClose(AbstractStructSchema<Cpu> schema) throws Exception {
         // test open
         Map<String, Object> map = new HashMap();
         map.put("influxdbUrl", "http://localhost:9999");

From 9f4a5faebec3b593d2b7a721a1af27c4c605dcf4 Mon Sep 17 00:00:00 2001
From: wangguowei <wangguowei@zhihu.com>
Date: Wed, 28 Oct 2020 11:08:56 +0800
Subject: [PATCH 03/10] fix bug

---
 .../impl/schema/generic/GenericProtobufNativeReaderTest.java  | 2 +-
 .../org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java    | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
index b51c3c62984fe..1dfac39e254a6 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
@@ -46,7 +46,7 @@ public void setup() {
     }
 
     @Test
-    public void testGenericAvroReaderByClazzBasedWriterSchema() {
+    public void testGenericReaderByClazzBasedWriterSchema() {
         message = TestMessage.newBuilder().setStringField(STRING_FIELD_VLUE).setDoubleField(DOUBLE_FIELD_VLUE).build();
         GenericProtobufNativeReader genericProtobufNativeReader = new GenericProtobufNativeReader(genericProtobufNativeSchema.getProtobufNativeSchema());
         GenericRecord genericRecordByWriterSchema = genericProtobufNativeReader.read(message.toByteArray());
diff --git a/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java b/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java
index 02919aa8aba41..a2e77cde5267f 100644
--- a/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java
+++ b/pulsar-io/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v2/InfluxDBSinkTest.java
@@ -31,7 +31,7 @@
 import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
 import org.apache.pulsar.client.impl.schema.AvroSchema;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
-import org.apache.pulsar.client.impl.schema.AbstractStructSchema;
+import org.apache.pulsar.client.impl.schema.StructSchema;
 import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
 import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
 import org.apache.pulsar.functions.api.Record;
@@ -144,7 +144,7 @@ public void testOpenWriteCloseJson() throws Exception {
         openWriteClose(jsonSchema);
     }
 
-    private void openWriteClose(AbstractStructSchema<Cpu> schema) throws Exception {
+    private void openWriteClose(StructSchema<Cpu> schema) throws Exception {
         // test open
         Map<String, Object> map = new HashMap();
         map.put("influxdbUrl", "http://localhost:9999");

From 9f98f5977ad040743e9d9a9838a0f6d090594ad5 Mon Sep 17 00:00:00 2001
From: wangguowei <wangguowei@zhihu.com>
Date: Wed, 28 Oct 2020 14:53:52 +0800
Subject: [PATCH 04/10]  fix License check error

---
 .../schema/ProtobufNativeSchemaCompatibilityCheck.java      | 6 +++---
 pulsar-broker/src/main/proto/SchemaRegistryFormat.proto     | 2 +-
 .../schema/ProtobufNativeSchemaCompatibilityCheckTest.java  | 6 +++---
 .../pulsar/client/impl/schema/ProtobufNativeSchema.java     | 6 +++---
 .../client/impl/schema/ProtobufNativeSchemaUtils.java       | 6 +++---
 .../client/impl/schema/reader/ProtobufNativeReader.java     | 6 +++---
 .../client/impl/schema/writer/ProtobufNativeWriter.java     | 6 +++---
 .../pulsar/client/impl/schema/ProtobufNativeSchemaTest.java | 6 +++---
 .../impl/schema/generic/AbstractGenericSchemaTest.java      | 6 +++---
 .../schema/generic/GenericProtobufNativeReaderTest.java     | 6 +++---
 .../schema/generic/GenericProtobufNativeSchemaTest.java     | 6 +++---
 11 files changed, 31 insertions(+), 31 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
index 4550a076284fa..1e2f8434b7721 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto b/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto
index 5bdec8b563b24..6ceb8ad3ada81 100644
--- a/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto
+++ b/pulsar-broker/src/main/proto/SchemaRegistryFormat.proto
@@ -1,4 +1,4 @@
-   /**
+/**
  * 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
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java
index 75f6b6dd76e04..2c5d099f551e0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheckTest.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java
index 2a46155797db9..343fb93743c02 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchema.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
index 3c8e51111f2d4..f01451f15f67b 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java
index 9701e2ffa1b13..1e75fccd71b04 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/reader/ProtobufNativeReader.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java
index 82f241c9370bd..44d9128a26cdd 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/ProtobufNativeWriter.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java
index f1b0b65d18992..9300ce5da34fd 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
index f2e59b978fa9e..cee67a715f015 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
index 1dfac39e254a6..985f37dff3097 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeReaderTest.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java
index ac903d15a6969..886d7f8105cb0 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchemaTest.java
@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *   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

From 4378645b78ee977edeb4cd8d299ea93bd78ff0ff Mon Sep 17 00:00:00 2001
From: hnail <wangguowei_1988@126.com>
Date: Fri, 13 Nov 2020 15:09:57 +0800
Subject: [PATCH 05/10] Update
 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java

Co-authored-by: Sijie Guo <sg@streamnative.io>
---
 .../schema/validator/ProtobufNativeSchemaDataValidator.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
index 5479f4845cda6..df6d4aa21d729 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
@@ -31,7 +31,7 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
         try {
             descriptor = ProtobufNativeSchemaUtils.deserialize(schemaData.getData());
         } catch (Exception e) {
-            throw new InvalidSchemaDataException("deserialize  ProtobufNative Schema failed", e);
+            throw new InvalidSchemaDataException("deserialize ProtobufNative Schema failed", e);
         }
         if (descriptor == null) {
             throw new InvalidSchemaDataException("protobuf root message Descriptor is null , please recheck rootMessageTypeName or rootFileDescriptorName conf. ");

From 2487bf25ef6a1d2dd24796d963396fcc02aaf46a Mon Sep 17 00:00:00 2001
From: hnail <wangguowei_1988@126.com>
Date: Fri, 13 Nov 2020 15:10:12 +0800
Subject: [PATCH 06/10] Update
 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java

Co-authored-by: Sijie Guo <sg@streamnative.io>
---
 .../schema/validator/ProtobufNativeSchemaDataValidator.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
index df6d4aa21d729..acf2878f13a97 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/ProtobufNativeSchemaDataValidator.java
@@ -34,7 +34,7 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
             throw new InvalidSchemaDataException("deserialize ProtobufNative Schema failed", e);
         }
         if (descriptor == null) {
-            throw new InvalidSchemaDataException("protobuf root message Descriptor is null , please recheck rootMessageTypeName or rootFileDescriptorName conf. ");
+            throw new InvalidSchemaDataException("protobuf root message descriptor is null , please recheck rootMessageTypeName or rootFileDescriptorName conf. ");
         }
     }
 

From bb45c1469ff5bfa004c0d20d5279669cd26277bf Mon Sep 17 00:00:00 2001
From: hnail <wangguowei_1988@126.com>
Date: Fri, 13 Nov 2020 15:10:31 +0800
Subject: [PATCH 07/10] Update
 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java

Co-authored-by: congbo <39078850+congbobo184@users.noreply.github.com>
---
 .../src/main/java/org/apache/pulsar/client/api/Schema.java      | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
index 36f099325a2a4..b40a54b6e1e53 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
@@ -257,7 +257,7 @@ static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF(Sch
      * @param clazz the Protobuf generated class to be used to extract the schema
      * @return a Schema instance
      */
-    static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUFNATIVE(Class<T> clazz) {
+    static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF_NATIVE(Class<T> clazz) {
         return DefaultImplementation.newProtobufNativeSchema(SchemaDefinition.builder().withPojo(clazz).build());
     }
 

From 6a55bcc897ec45cf6ef436254b53e678d7eca36e Mon Sep 17 00:00:00 2001
From: hnail <wangguowei_1988@126.com>
Date: Fri, 13 Nov 2020 15:10:40 +0800
Subject: [PATCH 08/10] Update
 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java

Co-authored-by: congbo <39078850+congbobo184@users.noreply.github.com>
---
 .../src/main/java/org/apache/pulsar/client/api/Schema.java      | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
index b40a54b6e1e53..d66779fcdf956 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
@@ -267,7 +267,7 @@ static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF_NAT
      * @param schemaDefinition schemaDefinition the definition of the schema
      * @return a Schema instance
      */
-    static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUFNATIVE(
+    static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF_NATIVE(
             SchemaDefinition<T> schemaDefinition) {
         return DefaultImplementation.newProtobufNativeSchema(schemaDefinition);
     }

From 38e470f68cf89f4959acb5033397c97099b0d17a Mon Sep 17 00:00:00 2001
From: wangguowei <wangguowei@zhihu.com>
Date: Fri, 13 Nov 2020 16:50:18 +0800
Subject: [PATCH 09/10] codeStyle fix

---
 ...rotobufNativeSchemaCompatibilityCheck.java |  7 ++--
 .../schema/ProtobufNativeSchemaUtils.java     | 37 +++++++++++--------
 .../generic/GenericProtobufNativeSchema.java  |  9 ++++-
 .../generic/AbstractGenericSchemaTest.java    |  8 ++--
 4 files changed, 37 insertions(+), 24 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
index 1e2f8434b7721..6a3f7e65905b0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
@@ -47,12 +47,12 @@ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityS
             case FORWARD:
             case FULL_TRANSITIVE:
             case FULL:
-                CheckRootRootMessageChange(fromDescriptor, toDescriptor, strategy);
+                checkRootMessageChange(fromDescriptor, toDescriptor, strategy);
                 return;
             case ALWAYS_COMPATIBLE:
                 return;
             default:
-                throw new IncompatibleSchemaException("Unknown SchemaCompatibilityStrategy");
+                throw new IncompatibleSchemaException("Unknown SchemaCompatibilityStrategy.");
         }
     }
 
@@ -63,7 +63,8 @@ public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaComp
         }
     }
 
-    private void CheckRootRootMessageChange(Descriptor fromDescriptor, Descriptor toDescriptor, SchemaCompatibilityStrategy strategy) throws IncompatibleSchemaException {
+    private void checkRootMessageChange(Descriptor fromDescriptor, Descriptor toDescriptor,
+                                            SchemaCompatibilityStrategy strategy) throws IncompatibleSchemaException {
         if (!fromDescriptor.getFullName().equals(toDescriptor.getFullName())) {
             throw new IncompatibleSchemaException("Protobuf root message isn't allow change!");
         }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
index f01451f15f67b..5fe94ce259956 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
@@ -18,7 +18,11 @@
  */
 package org.apache.pulsar.client.impl.schema;
 
+import static com.google.protobuf.DescriptorProtos.FileDescriptorProto;
+import static com.google.protobuf.DescriptorProtos.FileDescriptorSet;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.protobuf.Descriptors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.SchemaSerializationException;
 import org.apache.pulsar.common.protocol.schema.ProtobufNativeSchemaData;
@@ -28,13 +32,14 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import static com.google.protobuf.DescriptorProtos.FileDescriptorProto;
-import static com.google.protobuf.DescriptorProtos.FileDescriptorSet;
-import static com.google.protobuf.Descriptors.*;
-
+/**
+ * Protobuf-Native schema util used for serialize and deserialize
+ * between {@link com.google.protobuf.Descriptors.Descriptor} and
+ * {@link org.apache.pulsar.common.protocol.schema.ProtobufNativeSchemaData}.
+ */
 public class ProtobufNativeSchemaUtils {
 
-    public static byte[] serialize(Descriptor descriptor) {
+    public static byte[] serialize(Descriptors.Descriptor descriptor) {
         byte[] schemaDataBytes;
         try {
             Map<String, FileDescriptorProto> fileDescriptorProtoCache = new HashMap<>();
@@ -59,7 +64,7 @@ public static byte[] serialize(Descriptor descriptor) {
         return schemaDataBytes;
     }
 
-    private static void serializeFileDescriptor(FileDescriptor fileDescriptor, Map<String, FileDescriptorProto> fileDescriptorCache) {
+    private static void serializeFileDescriptor(Descriptors.FileDescriptor fileDescriptor, Map<String, FileDescriptorProto> fileDescriptorCache) {
         fileDescriptor.getDependencies().forEach(dependency -> {
                     if (!fileDescriptorCache.containsKey(dependency.getFullName())) {
                         serializeFileDescriptor(dependency, fileDescriptorCache);
@@ -67,7 +72,7 @@ private static void serializeFileDescriptor(FileDescriptor fileDescriptor, Map<S
                 }
         );
         String[] unResolvedFileDescriptNames = fileDescriptor.getDependencies().stream().
-                filter(item -> !fileDescriptorCache.containsKey(item.getFullName())).map(FileDescriptor::getFullName).toArray(String[]::new);
+                filter(item -> !fileDescriptorCache.containsKey(item.getFullName())).map(Descriptors.FileDescriptor::getFullName).toArray(String[]::new);
         if (unResolvedFileDescriptNames.length == 0) {
             fileDescriptorCache.put(fileDescriptor.getFullName(), fileDescriptor.toProto());
         } else {
@@ -75,13 +80,13 @@ private static void serializeFileDescriptor(FileDescriptor fileDescriptor, Map<S
         }
     }
 
-    public static Descriptor deserialize(byte[] schemaDataBytes) {
-        Descriptor descriptor;
+    public static Descriptors.Descriptor deserialize(byte[] schemaDataBytes) {
+        Descriptors.Descriptor descriptor;
         try {
             ProtobufNativeSchemaData schemaData = new ObjectMapper().readValue(schemaDataBytes, ProtobufNativeSchemaData.class);
 
             Map<String, FileDescriptorProto> fileDescriptorProtoCache = new HashMap<>();
-            Map<String, FileDescriptor> fileDescriptorCache = new HashMap<>();
+            Map<String, Descriptors.FileDescriptor> fileDescriptorCache = new HashMap<>();
             FileDescriptorSet fileDescriptorSet = FileDescriptorSet.parseFrom(schemaData.getFileDescriptorSet());
             fileDescriptorSet.getFileList().forEach(fileDescriptorProto -> {
                 fileDescriptorProtoCache.put(fileDescriptorProto.getName(), fileDescriptorProto);
@@ -91,7 +96,7 @@ public static Descriptor deserialize(byte[] schemaDataBytes) {
             //recursively build FileDescriptor
             deserializeFileDescriptor(rootFileDescriptorProto, fileDescriptorCache, fileDescriptorProtoCache);
             //extract root fileDescriptor
-            FileDescriptor fileDescriptor = fileDescriptorCache.get(schemaData.getRootFileDescriptorName());
+            Descriptors.FileDescriptor fileDescriptor = fileDescriptorCache.get(schemaData.getRootFileDescriptorName());
             //trim package
             String[] paths = StringUtils.removeFirst(schemaData.getRootMessageTypeName(), fileDescriptor.getPackage()).replaceFirst(".", "").split("\\.");
             //extract root message
@@ -109,7 +114,7 @@ public static Descriptor deserialize(byte[] schemaDataBytes) {
         return descriptor;
     }
 
-    private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptorProto, Map<String, FileDescriptor> fileDescriptorCache, Map<String, FileDescriptorProto> fileDescriptorProtoCache) {
+    private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptorProto, Map<String, Descriptors.FileDescriptor> fileDescriptorCache, Map<String, FileDescriptorProto> fileDescriptorProtoCache) {
         fileDescriptorProto.getDependencyList().forEach(dependencyFileDescriptorName -> {
             if (!fileDescriptorCache.containsKey(dependencyFileDescriptorName)) {
                 FileDescriptorProto dependencyFileDescriptor = fileDescriptorProtoCache.get(dependencyFileDescriptorName);
@@ -117,18 +122,18 @@ private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptor
             }
         });
 
-        FileDescriptor[] dependencyFileDescriptors = fileDescriptorProto.getDependencyList().stream().map(dependency -> {
+        Descriptors.FileDescriptor[] dependencyFileDescriptors = fileDescriptorProto.getDependencyList().stream().map(dependency -> {
             if (fileDescriptorCache.containsKey(dependency)) {
                 return fileDescriptorCache.get(dependency);
             } else {
                 throw new SchemaSerializationException(fileDescriptorProto.getName() + "can't resolve  dependency '" + dependency + "'");
             }
-        }).toArray(FileDescriptor[]::new);
+        }).toArray(Descriptors.FileDescriptor[]::new);
 
         try {
-            FileDescriptor fileDescriptor = FileDescriptor.buildFrom(fileDescriptorProto, dependencyFileDescriptors);
+            Descriptors.FileDescriptor fileDescriptor = Descriptors.FileDescriptor.buildFrom(fileDescriptorProto, dependencyFileDescriptors);
             fileDescriptorCache.put(fileDescriptor.getFullName(), fileDescriptor);
-        } catch (DescriptorValidationException e) {
+        } catch (Descriptors.DescriptorValidationException e) {
             e.printStackTrace();
             throw new SchemaSerializationException(e);
         }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java
index 767350c0e4725..949ad9e2fcf78 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericProtobufNativeSchema.java
@@ -20,7 +20,11 @@
 
 import com.google.protobuf.Descriptors;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.client.api.schema.*;
+import org.apache.pulsar.client.api.schema.Field;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericRecordBuilder;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.SchemaReader;
 import org.apache.pulsar.client.impl.schema.ProtobufNativeSchemaUtils;
 import org.apache.pulsar.client.impl.schema.SchemaUtils;
 import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion;
@@ -29,6 +33,9 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+/**
+ * Generic ProtobufNative schema.
+ */
 @Slf4j
 public class GenericProtobufNativeSchema extends AbstractGenericSchema {
 
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
index cee67a715f015..0865579ffedd4 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/AbstractGenericSchemaTest.java
@@ -32,14 +32,14 @@
 import static org.testng.Assert.assertEquals;
 
 /**
- * Unit testing AbstractGenericSchema for non-avroBasedGenericSchema
+ * Unit testing AbstractGenericSchema for non-avroBasedGenericSchema.
  */
 @Slf4j
 public class AbstractGenericSchemaTest {
 
     @Test
     public void testGenericProtobufNativeSchema() {
-        Schema<org.apache.pulsar.client.schema.proto.Test.TestMessage> encodeSchema = Schema.PROTOBUFNATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
+        Schema<org.apache.pulsar.client.schema.proto.Test.TestMessage> encodeSchema = Schema.PROTOBUF_NATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
         GenericSchema decodeSchema = GenericProtobufNativeSchema.of(encodeSchema.getSchemaInfo());
 
         testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema);
@@ -49,12 +49,12 @@ public void testGenericProtobufNativeSchema() {
     public void testAutoProtobufNativeSchema() {
         // configure the schema info provider
         MultiVersionSchemaInfoProvider multiVersionSchemaInfoProvider = mock(MultiVersionSchemaInfoProvider.class);
-        GenericSchema genericProtobufNativeSchema = GenericProtobufNativeSchema.of(Schema.PROTOBUFNATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class).getSchemaInfo());
+        GenericSchema genericProtobufNativeSchema = GenericProtobufNativeSchema.of(Schema.PROTOBUF_NATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class).getSchemaInfo());
         when(multiVersionSchemaInfoProvider.getSchemaByVersion(any(byte[].class)))
                 .thenReturn(CompletableFuture.completedFuture(genericProtobufNativeSchema.getSchemaInfo()));
 
         // configure encode schema
-        Schema<org.apache.pulsar.client.schema.proto.Test.TestMessage> encodeSchema = Schema.PROTOBUFNATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
+        Schema<org.apache.pulsar.client.schema.proto.Test.TestMessage> encodeSchema = Schema.PROTOBUF_NATIVE(org.apache.pulsar.client.schema.proto.Test.TestMessage.class);
         // configure decode schema
         AutoConsumeSchema decodeSchema = new AutoConsumeSchema();
         decodeSchema.configureSchemaInfo("test-topic", "topic", encodeSchema.getSchemaInfo());

From 2597f2c6287783ba285737a28cb39cf3e058aa37 Mon Sep 17 00:00:00 2001
From: wangguowei <wangguowei@zhihu.com>
Date: Mon, 16 Nov 2020 14:07:43 +0800
Subject: [PATCH 10/10] codeStyle fix v2

---
 .../impl/schema/ProtobufNativeSchemaUtils.java     | 14 ++++++--------
 1 file changed, 6 insertions(+), 8 deletions(-)

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
index 5fe94ce259956..7a806b5891588 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java
@@ -33,7 +33,7 @@
 import java.util.Map;
 
 /**
- * Protobuf-Native schema util used for serialize and deserialize
+ * Protobuf-Native schema util used for serialize/deserialize
  * between {@link com.google.protobuf.Descriptors.Descriptor} and
  * {@link org.apache.pulsar.common.protocol.schema.ProtobufNativeSchemaData}.
  */
@@ -56,7 +56,7 @@ public static byte[] serialize(Descriptors.Descriptor descriptor) {
             ProtobufNativeSchemaData schemaData = ProtobufNativeSchemaData.builder().fileDescriptorSet(fileDescriptorSet)
                     .rootFileDescriptorName(rootFileDescriptorName).rootMessageTypeName(rootMessageTypeName).build();
             schemaDataBytes = new ObjectMapper().writeValueAsBytes(schemaData);
-            logger.debug("descriptor : {} serialized to : {} ", descriptor.getFullName(), schemaDataBytes);
+            logger.debug("descriptor '{}' serialized to '{}'.", descriptor.getFullName(), schemaDataBytes);
         } catch (Exception e) {
             e.printStackTrace();
             throw new SchemaSerializationException(e);
@@ -76,7 +76,7 @@ private static void serializeFileDescriptor(Descriptors.FileDescriptor fileDescr
         if (unResolvedFileDescriptNames.length == 0) {
             fileDescriptorCache.put(fileDescriptor.getFullName(), fileDescriptor.toProto());
         } else {
-            throw new SchemaSerializationException(fileDescriptor.getFullName() + " can't resolve dependency :" + unResolvedFileDescriptNames);
+            throw new SchemaSerializationException(fileDescriptor.getFullName() + " can't resolve dependency '" + unResolvedFileDescriptNames + "'.");
         }
     }
 
@@ -88,9 +88,7 @@ public static Descriptors.Descriptor deserialize(byte[] schemaDataBytes) {
             Map<String, FileDescriptorProto> fileDescriptorProtoCache = new HashMap<>();
             Map<String, Descriptors.FileDescriptor> fileDescriptorCache = new HashMap<>();
             FileDescriptorSet fileDescriptorSet = FileDescriptorSet.parseFrom(schemaData.getFileDescriptorSet());
-            fileDescriptorSet.getFileList().forEach(fileDescriptorProto -> {
-                fileDescriptorProtoCache.put(fileDescriptorProto.getName(), fileDescriptorProto);
-            });
+            fileDescriptorSet.getFileList().forEach(fileDescriptorProto -> fileDescriptorProtoCache.put(fileDescriptorProto.getName(), fileDescriptorProto));
             FileDescriptorProto rootFileDescriptorProto = fileDescriptorProtoCache.get(schemaData.getRootFileDescriptorName());
 
             //recursively build FileDescriptor
@@ -105,7 +103,7 @@ public static Descriptors.Descriptor deserialize(byte[] schemaDataBytes) {
             for (int i = 1; i < paths.length; i++) {
                 descriptor = descriptor.findNestedTypeByName(paths[i]);
             }
-            logger.debug("deserialize : {} serialized to descriptor: {} ", schemaDataBytes, descriptor.getFullName());
+            logger.debug("deserialize '{}' to descriptor: '{}'.", schemaDataBytes, descriptor.getFullName());
         } catch (Exception e) {
             e.printStackTrace();
             throw new SchemaSerializationException(e);
@@ -126,7 +124,7 @@ private static void deserializeFileDescriptor(FileDescriptorProto fileDescriptor
             if (fileDescriptorCache.containsKey(dependency)) {
                 return fileDescriptorCache.get(dependency);
             } else {
-                throw new SchemaSerializationException(fileDescriptorProto.getName() + "can't resolve  dependency '" + dependency + "'");
+                throw new SchemaSerializationException("'" + fileDescriptorProto.getName() + "' can't resolve  dependency '" + dependency + "'.");
             }
         }).toArray(Descriptors.FileDescriptor[]::new);