diff --git a/build/docker/Dockerfile b/build/docker/Dockerfile
index ac973d0a6b1cb..4046c25768f05 100644
--- a/build/docker/Dockerfile
+++ b/build/docker/Dockerfile
@@ -21,19 +21,27 @@ FROM ubuntu:16.04
 
 # prepare the directory for pulsar related files
 RUN mkdir /pulsar
-ADD protobuf.patch /pulsar
 
 RUN apt-get update && \
     apt-get install -y software-properties-common && \
     add-apt-repository ppa:openjdk-r/ppa && \
     apt-get update && \
     apt-get install -y tig g++ cmake libssl-dev libcurl4-openssl-dev \
-                liblog4cxx-dev libprotobuf-dev google-mock libgtest-dev \
+                liblog4cxx-dev google-mock libgtest-dev \
                 libboost-dev libboost-program-options-dev libboost-system-dev libboost-python-dev \
                 libxml2-utils protobuf-compiler wget \
                 curl doxygen openjdk-8-jdk-headless openjdk-11-jdk-headless clang-format-5.0 \
                 gnupg2 golang-1.13-go zip unzip libzstd-dev libsnappy-dev python3-pip libpython-dev
 
+# Build protobuf 3.x.y from source since the default protobuf from Ubuntu's apt source is 2.x.y
+RUN curl -O -L https://github.com/protocolbuffers/protobuf/releases/download/v3.17.3/protobuf-cpp-3.17.3.tar.gz && \
+    tar xvfz protobuf-cpp-3.17.3.tar.gz && \
+    cd protobuf-3.17.3/ && \
+    CXXFLAGS=-fPIC ./configure && \
+    make -j8 && make install && \
+    cd .. && rm -rf protobuf-3.17.3/ protobuf-cpp-3.17.3.tar.gz
+ENV LD_LIBRARY_PATH /usr/local/lib
+
 # Compile and install gtest
 RUN cd /usr/src/gtest && cmake . && make && cp libgtest.a /usr/lib
 
@@ -49,8 +57,8 @@ ENV JAVA_HOME_11=/usr/lib/jvm/java-1.11.0-openjdk-amd64
 ## Website build dependencies
 
 # Install Ruby-2.4.1
-RUN apt-get install -y
-RUN gpg2 --keyserver hkp://keys.gnupg.net --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB && \
+RUN (curl -sSL https://rvm.io/mpapis.asc | gpg --import -) && \
+    (curl -sSL https://rvm.io/pkuczynski.asc | gpg --import -) && \
     (curl -sSL https://get.rvm.io | bash -s stable)
 ENV PATH "$PATH:/usr/local/rvm/bin"
 RUN rvm install 2.4.1
@@ -70,20 +78,6 @@ RUN dpkg -i crowdin.deb
 RUN wget https://bootstrap.pypa.io/pip/2.7/get-pip.py && python get-pip.py && rm get-pip.py
 RUN pip3 install pdoc
 
-# Install Protobuf doc generator (requires Go)
-ENV GOPATH "$HOME/go"
-ENV PATH "/usr/lib/go-1.13/bin:$GOPATH/bin:$PATH"
-RUN go get -u github.com/pseudomuto/protoc-gen-doc/cmd/protoc-gen-doc
-
-# Build the patched protoc
-RUN git clone https://github.com/google/protobuf.git /pulsar/protobuf && \
-    cd /pulsar/protobuf && \
-    git checkout v2.4.1 && \
-    patch -p1 < /pulsar/protobuf.patch && \
-    autoreconf --install && \
-    ./configure && \
-    make
-
 # Installation
 ARG MAVEN_VERSION=3.6.3
 ARG MAVEN_FILENAME="apache-maven-${MAVEN_VERSION}-bin.tar.gz"
diff --git a/build/docker/protobuf.patch b/build/docker/protobuf.patch
deleted file mode 100644
index fe8bf669afdd4..0000000000000
--- a/build/docker/protobuf.patch
+++ /dev/null
@@ -1,1170 +0,0 @@
-diff --git a/src/google/protobuf/compiler/java/java_enum.cc b/src/google/protobuf/compiler/java/java_enum.cc
-index 9d7bcab6..71c4ab70 100644
---- a/src/google/protobuf/compiler/java/java_enum.cc
-+++ b/src/google/protobuf/compiler/java/java_enum.cc
-@@ -70,12 +70,12 @@ void EnumGenerator::Generate(io::Printer* printer) {
-   if (HasDescriptorMethods(descriptor_)) {
-     printer->Print(
-       "public enum $classname$\n"
--      "    implements com.google.protobuf.ProtocolMessageEnum {\n",
-+      "    implements org.apache.pulsar.shaded.com.google.protobuf.v241.ProtocolMessageEnum {\n",
-       "classname", descriptor_->name());
-   } else {
-     printer->Print(
-       "public enum $classname$\n"
--      "    implements com.google.protobuf.Internal.EnumLite {\n",
-+      "    implements org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLite {\n",
-       "classname", descriptor_->name());
-   }
-   printer->Indent();
-@@ -139,13 +139,13 @@ void EnumGenerator::Generate(io::Printer* printer) {
-     "  }\n"
-     "}\n"
-     "\n"
--    "public static com.google.protobuf.Internal.EnumLiteMap<$classname$>\n"
-+    "public static org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLiteMap<$classname$>\n"
-     "    internalGetValueMap() {\n"
-     "  return internalValueMap;\n"
-     "}\n"
--    "private static com.google.protobuf.Internal.EnumLiteMap<$classname$>\n"
-+    "private static org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLiteMap<$classname$>\n"
-     "    internalValueMap =\n"
--    "      new com.google.protobuf.Internal.EnumLiteMap<$classname$>() {\n"
-+    "      new org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.EnumLiteMap<$classname$>() {\n"
-     "        public $classname$ findValueByNumber(int number) {\n"
-     "          return $classname$.valueOf(number);\n"
-     "        }\n"
-@@ -158,15 +158,15 @@ void EnumGenerator::Generate(io::Printer* printer) {
- 
-   if (HasDescriptorMethods(descriptor_)) {
-     printer->Print(
--      "public final com.google.protobuf.Descriptors.EnumValueDescriptor\n"
-+      "public final org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.EnumValueDescriptor\n"
-       "    getValueDescriptor() {\n"
-       "  return getDescriptor().getValues().get(index);\n"
-       "}\n"
--      "public final com.google.protobuf.Descriptors.EnumDescriptor\n"
-+      "public final org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.EnumDescriptor\n"
-       "    getDescriptorForType() {\n"
-       "  return getDescriptor();\n"
-       "}\n"
--      "public static final com.google.protobuf.Descriptors.EnumDescriptor\n"
-+      "public static final org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.EnumDescriptor\n"
-       "    getDescriptor() {\n");
- 
-     // TODO(kenton):  Cache statically?  Note that we can't access descriptors
-@@ -201,7 +201,7 @@ void EnumGenerator::Generate(io::Printer* printer) {
-       "};\n"
-       "\n"
-       "public static $classname$ valueOf(\n"
--      "    com.google.protobuf.Descriptors.EnumValueDescriptor desc) {\n"
-+      "    org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.EnumValueDescriptor desc) {\n"
-       "  if (desc.getType() != getDescriptor()) {\n"
-       "    throw new java.lang.IllegalArgumentException(\n"
-       "      \"EnumValueDescriptor is not for this type.\");\n"
-diff --git a/src/google/protobuf/compiler/java/java_enum_field.cc b/src/google/protobuf/compiler/java/java_enum_field.cc
-index 72caa10b..fb7ef574 100644
---- a/src/google/protobuf/compiler/java/java_enum_field.cc
-+++ b/src/google/protobuf/compiler/java/java_enum_field.cc
-@@ -227,7 +227,7 @@ void EnumFieldGenerator::
- GenerateSerializedSizeCode(io::Printer* printer) const {
-   printer->Print(variables_,
-     "if ($get_has_field_bit_message$) {\n"
--    "  size += com.google.protobuf.CodedOutputStream\n"
-+    "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-     "    .computeEnumSize($number$, $name$_.getNumber());\n"
-     "}\n");
- }
-@@ -484,7 +484,7 @@ GenerateSerializedSizeCode(io::Printer* printer) const {
- 
-   printer->Print(variables_,
-     "for (int i = 0; i < $name$_.size(); i++) {\n"
--    "  dataSize += com.google.protobuf.CodedOutputStream\n"
-+    "  dataSize += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-     "    .computeEnumSizeNoTag($name$_.get(i).getNumber());\n"
-     "}\n");
-   printer->Print(
-@@ -493,7 +493,7 @@ GenerateSerializedSizeCode(io::Printer* printer) const {
-     printer->Print(variables_,
-       "if (!get$capitalized_name$List().isEmpty()) {"
-       "  size += $tag_size$;\n"
--      "  size += com.google.protobuf.CodedOutputStream\n"
-+      "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-       "    .computeRawVarint32Size(dataSize);\n"
-       "}");
-   } else {
-diff --git a/src/google/protobuf/compiler/java/java_extension.cc b/src/google/protobuf/compiler/java/java_extension.cc
-index 9b147c77..99846474 100644
---- a/src/google/protobuf/compiler/java/java_extension.cc
-+++ b/src/google/protobuf/compiler/java/java_extension.cc
-@@ -137,9 +137,9 @@ void ExtensionGenerator::Generate(io::Printer* printer) {
-       printer->Print(
-           vars,
-           "public static final\n"
--          "  com.google.protobuf.GeneratedMessage.GeneratedExtension<\n"
-+          "  org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.GeneratedExtension<\n"
-           "    $containing_type$,\n"
--          "    $type$> $name$ = com.google.protobuf.GeneratedMessage\n"
-+          "    $type$> $name$ = org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage\n"
-           "        .newFileScopedGeneratedExtension(\n"
-           "      $singular_type$.class,\n"
-           "      $prototype$);\n");
-@@ -148,9 +148,9 @@ void ExtensionGenerator::Generate(io::Printer* printer) {
-       printer->Print(
-           vars,
-           "public static final\n"
--          "  com.google.protobuf.GeneratedMessage.GeneratedExtension<\n"
-+          "  org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.GeneratedExtension<\n"
-           "    $containing_type$,\n"
--          "    $type$> $name$ = com.google.protobuf.GeneratedMessage\n"
-+          "    $type$> $name$ = org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage\n"
-           "        .newMessageScopedGeneratedExtension(\n"
-           "      $scope$.getDefaultInstance(),\n"
-           "      $index$,\n"
-@@ -163,30 +163,30 @@ void ExtensionGenerator::Generate(io::Printer* printer) {
-       printer->Print(
-           vars,
-           "public static final\n"
--          "  com.google.protobuf.GeneratedMessageLite.GeneratedExtension<\n"
-+          "  org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite.GeneratedExtension<\n"
-           "    $containing_type$,\n"
--          "    $type$> $name$ = com.google.protobuf.GeneratedMessageLite\n"
-+          "    $type$> $name$ = org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite\n"
-           "        .newRepeatedGeneratedExtension(\n"
-           "      $containing_type$.getDefaultInstance(),\n"
-           "      $prototype$,\n"
-           "      $enum_map$,\n"
-           "      $number$,\n"
--          "      com.google.protobuf.WireFormat.FieldType.$type_constant$,\n"
-+          "      org.apache.pulsar.shaded.com.google.protobuf.v241.WireFormat.FieldType.$type_constant$,\n"
-           "      $packed$);\n");
-     } else {
-       printer->Print(
-           vars,
-           "public static final\n"
--          "  com.google.protobuf.GeneratedMessageLite.GeneratedExtension<\n"
-+          "  org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite.GeneratedExtension<\n"
-           "    $containing_type$,\n"
--          "    $type$> $name$ = com.google.protobuf.GeneratedMessageLite\n"
-+          "    $type$> $name$ = org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite\n"
-           "        .newSingularGeneratedExtension(\n"
-           "      $containing_type$.getDefaultInstance(),\n"
-           "      $default$,\n"
-           "      $prototype$,\n"
-           "      $enum_map$,\n"
-           "      $number$,\n"
--          "      com.google.protobuf.WireFormat.FieldType.$type_constant$);\n");
-+          "      org.apache.pulsar.shaded.com.google.protobuf.v241.WireFormat.FieldType.$type_constant$);\n");
-     }
-   }
- }
-diff --git a/src/google/protobuf/compiler/java/java_file.cc b/src/google/protobuf/compiler/java/java_file.cc
-index 8968069f..86620b74 100644
---- a/src/google/protobuf/compiler/java/java_file.cc
-+++ b/src/google/protobuf/compiler/java/java_file.cc
-@@ -155,7 +155,7 @@ void FileGenerator::Generate(io::Printer* printer) {
- 
-   printer->Print(
-     "public static void registerAllExtensions(\n"
--    "    com.google.protobuf.ExtensionRegistry$lite$ registry) {\n",
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistry$lite$ registry) {\n",
-     "lite", HasDescriptorMethods(file_) ? "" : "Lite");
- 
-   printer->Indent();
-@@ -249,11 +249,11 @@ void FileGenerator::GenerateEmbeddedDescriptor(io::Printer* printer) {
-   file_proto.SerializeToString(&file_data);
- 
-   printer->Print(
--    "public static com.google.protobuf.Descriptors.FileDescriptor\n"
-+    "public static org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.FileDescriptor\n"
-     "    getDescriptor() {\n"
-     "  return descriptor;\n"
-     "}\n"
--    "private static com.google.protobuf.Descriptors.FileDescriptor\n"
-+    "private static org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.FileDescriptor\n"
-     "    descriptor;\n"
-     "static {\n"
-     "  java.lang.String[] descriptorData = {\n");
-@@ -283,12 +283,12 @@ void FileGenerator::GenerateEmbeddedDescriptor(io::Printer* printer) {
-   // Create the InternalDescriptorAssigner.
- 
-   printer->Print(
--    "com.google.protobuf.Descriptors.FileDescriptor."
-+    "org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.FileDescriptor."
-       "InternalDescriptorAssigner assigner =\n"
--    "  new com.google.protobuf.Descriptors.FileDescriptor."
-+    "  new org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.FileDescriptor."
-       "InternalDescriptorAssigner() {\n"
--    "    public com.google.protobuf.ExtensionRegistry assignDescriptors(\n"
--    "        com.google.protobuf.Descriptors.FileDescriptor root) {\n"
-+    "    public org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistry assignDescriptors(\n"
-+    "        org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.FileDescriptor root) {\n"
-     "      descriptor = root;\n");
- 
-   printer->Indent();
-@@ -310,8 +310,8 @@ void FileGenerator::GenerateEmbeddedDescriptor(io::Printer* printer) {
-     // Must construct an ExtensionRegistry containing all possible extensions
-     // and return it.
-     printer->Print(
--      "com.google.protobuf.ExtensionRegistry registry =\n"
--      "  com.google.protobuf.ExtensionRegistry.newInstance();\n"
-+      "org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistry registry =\n"
-+      "  org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistry.newInstance();\n"
-       "registerAllExtensions(registry);\n");
-     for (int i = 0; i < file_->dependency_count(); i++) {
-       if (ShouldIncludeDependency(file_->dependency(i))) {
-@@ -339,9 +339,9 @@ void FileGenerator::GenerateEmbeddedDescriptor(io::Printer* printer) {
-   // Invoke internalBuildGeneratedFileFrom() to build the file.
- 
-   printer->Print(
--    "com.google.protobuf.Descriptors.FileDescriptor\n"
-+    "org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.FileDescriptor\n"
-     "  .internalBuildGeneratedFileFrom(descriptorData,\n"
--    "    new com.google.protobuf.Descriptors.FileDescriptor[] {\n");
-+    "    new org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.FileDescriptor[] {\n");
- 
-   for (int i = 0; i < file_->dependency_count(); i++) {
-     if (ShouldIncludeDependency(file_->dependency(i))) {
-diff --git a/src/google/protobuf/compiler/java/java_helpers.cc b/src/google/protobuf/compiler/java/java_helpers.cc
-index 1b6f1653..25a6e300 100644
---- a/src/google/protobuf/compiler/java/java_helpers.cc
-+++ b/src/google/protobuf/compiler/java/java_helpers.cc
-@@ -248,7 +248,7 @@ const char* BoxedPrimitiveTypeName(JavaType type) {
-     case JAVATYPE_DOUBLE : return "java.lang.Double";
-     case JAVATYPE_BOOLEAN: return "java.lang.Boolean";
-     case JAVATYPE_STRING : return "java.lang.String";
--    case JAVATYPE_BYTES  : return "com.google.protobuf.ByteString";
-+    case JAVATYPE_BYTES  : return "org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString";
-     case JAVATYPE_ENUM   : return NULL;
-     case JAVATYPE_MESSAGE: return NULL;
- 
-@@ -314,10 +314,10 @@ string DefaultValue(const FieldDescriptor* field) {
-         if (field->has_default_value()) {
-           // See comments in Internal.java for gory details.
-           return strings::Substitute(
--            "com.google.protobuf.Internal.bytesDefaultValue(\"$0\")",
-+            "org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.bytesDefaultValue(\"$0\")",
-             CEscape(field->default_value_string()));
-         } else {
--          return "com.google.protobuf.ByteString.EMPTY";
-+          return "org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString.EMPTY";
-         }
-       } else {
-         if (AllAscii(field->default_value_string())) {
-@@ -326,7 +326,7 @@ string DefaultValue(const FieldDescriptor* field) {
-         } else {
-           // See comments in Internal.java for gory details.
-           return strings::Substitute(
--            "com.google.protobuf.Internal.stringDefaultValue(\"$0\")",
-+            "org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.stringDefaultValue(\"$0\")",
-             CEscape(field->default_value_string()));
-         }
-       }
-diff --git a/src/google/protobuf/compiler/java/java_message.cc b/src/google/protobuf/compiler/java/java_message.cc
-index 4c087db5..3ae1c14c 100644
---- a/src/google/protobuf/compiler/java/java_message.cc
-+++ b/src/google/protobuf/compiler/java/java_message.cc
-@@ -155,7 +155,7 @@ MessageGenerator::~MessageGenerator() {}
- 
- void MessageGenerator::GenerateStaticVariables(io::Printer* printer) {
-   if (HasDescriptorMethods(descriptor_)) {
--    // Because descriptor.proto (com.google.protobuf.DescriptorProtos) is
-+    // Because descriptor.proto (org.apache.pulsar.shaded.com.google.protobuf.v241.DescriptorProtos) is
-     // used in the construction of descriptors, we have a tricky bootstrapping
-     // problem.  To help control static initialization order, we make sure all
-     // descriptors and other static data that depends on them are members of
-@@ -180,13 +180,13 @@ void MessageGenerator::GenerateStaticVariables(io::Printer* printer) {
- 
-     // The descriptor for this type.
-     printer->Print(vars,
--      "$private$static com.google.protobuf.Descriptors.Descriptor\n"
-+      "$private$static org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.Descriptor\n"
-       "  internal_$identifier$_descriptor;\n");
- 
-     // And the FieldAccessorTable.
-     printer->Print(vars,
-       "$private$static\n"
--      "  com.google.protobuf.GeneratedMessage.FieldAccessorTable\n"
-+      "  org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.FieldAccessorTable\n"
-       "    internal_$identifier$_fieldAccessorTable;\n");
-   }
- 
-@@ -224,7 +224,7 @@ void MessageGenerator::GenerateStaticVariableInitializers(
-     // And the FieldAccessorTable.
-     printer->Print(vars,
-       "internal_$identifier$_fieldAccessorTable = new\n"
--      "  com.google.protobuf.GeneratedMessage.FieldAccessorTable(\n"
-+      "  org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.FieldAccessorTable(\n"
-       "    internal_$identifier$_descriptor,\n"
-       "    new java.lang.String[] { ");
-     for (int i = 0; i < descriptor_->field_count(); i++) {
-@@ -255,13 +255,13 @@ void MessageGenerator::GenerateInterface(io::Printer* printer) {
-     if (HasDescriptorMethods(descriptor_)) {
-       printer->Print(
-         "public interface $classname$OrBuilder extends\n"
--        "    com.google.protobuf.GeneratedMessage.\n"
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.\n"
-         "        ExtendableMessageOrBuilder<$classname$> {\n",
-         "classname", descriptor_->name());
-     } else {
-       printer->Print(
-         "public interface $classname$OrBuilder extends \n"
--        "     com.google.protobuf.GeneratedMessageLite.\n"
-+        "     org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite.\n"
-         "          ExtendableMessageOrBuilder<$classname$> {\n",
-         "classname", descriptor_->name());
-     }
-@@ -269,12 +269,12 @@ void MessageGenerator::GenerateInterface(io::Printer* printer) {
-     if (HasDescriptorMethods(descriptor_)) {
-       printer->Print(
-         "public interface $classname$OrBuilder\n"
--        "    extends com.google.protobuf.MessageOrBuilder {\n",
-+        "    extends org.apache.pulsar.shaded.com.google.protobuf.v241.MessageOrBuilder {\n",
-         "classname", descriptor_->name());
-     } else {
-       printer->Print(
-         "public interface $classname$OrBuilder\n"
--        "    extends com.google.protobuf.MessageLiteOrBuilder {\n",
-+        "    extends org.apache.pulsar.shaded.com.google.protobuf.v241.MessageLiteOrBuilder {\n",
-         "classname", descriptor_->name());
-     }
-   }
-@@ -302,15 +302,15 @@ void MessageGenerator::Generate(io::Printer* printer) {
-     if (HasDescriptorMethods(descriptor_)) {
-       printer->Print(
-         "public $static$ final class $classname$ extends\n"
--        "    com.google.protobuf.GeneratedMessage.ExtendableMessage<\n"
--        "      $classname$> implements $classname$OrBuilder {\n",
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.ExtendableMessage<\n"
-+        "      $classname$> implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream.ByteBufGeneratedMessage {\n",
-         "static", is_own_file ? "" : "static",
-         "classname", descriptor_->name());
-     } else {
-       printer->Print(
-         "public $static$ final class $classname$ extends\n"
--        "    com.google.protobuf.GeneratedMessageLite.ExtendableMessage<\n"
--        "      $classname$> implements $classname$OrBuilder {\n",
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite.ExtendableMessage<\n"
-+        "      $classname$> implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream.ByteBufGeneratedMessage  {\n",
-         "static", is_own_file ? "" : "static",
-         "classname", descriptor_->name());
-     }
-@@ -318,25 +318,57 @@ void MessageGenerator::Generate(io::Printer* printer) {
-     if (HasDescriptorMethods(descriptor_)) {
-       printer->Print(
-         "public $static$ final class $classname$ extends\n"
--        "    com.google.protobuf.GeneratedMessage\n"
--        "    implements $classname$OrBuilder {\n",
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage\n"
-+        "    implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream.ByteBufGeneratedMessage  {\n",
-         "static", is_own_file ? "" : "static",
-         "classname", descriptor_->name());
-     } else {
-       printer->Print(
-         "public $static$ final class $classname$ extends\n"
--        "    com.google.protobuf.GeneratedMessageLite\n"
--        "    implements $classname$OrBuilder {\n",
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite\n"
-+        "    implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream.ByteBufGeneratedMessage  {\n",
-         "static", is_own_file ? "" : "static",
-         "classname", descriptor_->name());
-     }
-   }
-+
-+  // Integers for bit fields.
-+  int totalBits = 0;
-+  for (int i = 0; i < descriptor_->field_count(); i++) {
-+    totalBits += field_generators_.get(descriptor_->field(i))
-+        .GetNumBitsForMessage();
-+  }
-+  int totalInts = (totalBits + 31) / 32;
-+
-   printer->Indent();
-   printer->Print(
-     "// Use $classname$.newBuilder() to construct.\n"
--    "private $classname$(Builder builder) {\n"
--    "  super(builder);\n"
-+    "private io.netty.util.Recycler.Handle handle;\n"
-+    "private $classname$(io.netty.util.Recycler.Handle handle) {\n"
-+    "  this.handle = handle;\n"
-     "}\n"
-+    "\n"
-+    " private static final io.netty.util.Recycler<$classname$> RECYCLER = new io.netty.util.Recycler<$classname$>() {\n"
-+  "        protected $classname$ newObject(Handle handle) {\n"
-+  "          return new $classname$(handle);\n"
-+  "        }\n"
-+  "      };\n"
-+  "    \n"
-+  "    public void recycle() {\n"
-+  "        this.initFields();\n"
-+  "        this.memoizedIsInitialized = -1;\n",
-+    "classname", descriptor_->name());
-+
-+    for (int i = 0; i < totalInts; i++) {
-+      printer->Print("        this.$bit_field_name$ = 0;\n",
-+        "bit_field_name", GetBitFieldName(i));
-+    }
-+
-+    printer->Print(
-+  "        this.memoizedSerializedSize = -1;\n"
-+  "        if (handle != null) { RECYCLER.recycle(this, handle); }\n"
-+  "    }\n"
-+   "     \n"
-     // Used when constructing the default instance, which cannot be initialized
-     // immediately because it may cyclically refer to other default instances.
-     "private $classname$(boolean noInit) {}\n"
-@@ -365,13 +397,6 @@ void MessageGenerator::Generate(io::Printer* printer) {
-     messageGenerator.Generate(printer);
-   }
- 
--  // Integers for bit fields.
--  int totalBits = 0;
--  for (int i = 0; i < descriptor_->field_count(); i++) {
--    totalBits += field_generators_.get(descriptor_->field(i))
--        .GetNumBitsForMessage();
--  }
--  int totalInts = (totalBits + 31) / 32;
-   for (int i = 0; i < totalInts; i++) {
-     printer->Print("private int $bit_field_name$;\n",
-       "bit_field_name", GetBitFieldName(i));
-@@ -450,7 +475,12 @@ GenerateMessageSerializationMethods(io::Printer* printer) {
-        ExtensionRangeOrdering());
- 
-   printer->Print(
--    "public void writeTo(com.google.protobuf.CodedOutputStream output)\n"
-+      "public void writeTo(org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream output)\n"
-+      "                    throws java.io.IOException {\n"
-+          "    throw new RuntimeException(\"Cannot use CodedOutputStream\");\n"
-+          "}\n\n"
-+
-+    "public void writeTo(org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream output)\n"
-     "                    throws java.io.IOException {\n");
-   printer->Indent();
-   // writeTo(CodedOutputStream output) might be invoked without
-@@ -465,14 +495,14 @@ GenerateMessageSerializationMethods(io::Printer* printer) {
-   if (descriptor_->extension_range_count() > 0) {
-     if (descriptor_->options().message_set_wire_format()) {
-       printer->Print(
--        "com.google.protobuf.GeneratedMessage$lite$\n"
-+        "org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage$lite$\n"
-         "  .ExtendableMessage<$classname$>.ExtensionWriter extensionWriter =\n"
-         "    newMessageSetExtensionWriter();\n",
-         "lite", HasDescriptorMethods(descriptor_) ? "" : "Lite",
-         "classname", ClassName(descriptor_));
-     } else {
-       printer->Print(
--        "com.google.protobuf.GeneratedMessage$lite$\n"
-+        "org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage$lite$\n"
-         "  .ExtendableMessage<$classname$>.ExtensionWriter extensionWriter =\n"
-         "    newExtensionWriter();\n",
-         "lite", HasDescriptorMethods(descriptor_) ? "" : "Lite",
-@@ -565,25 +595,27 @@ GenerateParseFromMethods(io::Printer* printer) {
-   //   for code size.
-   printer->Print(
-     "public static $classname$ parseFrom(\n"
--    "    com.google.protobuf.ByteString data)\n"
--    "    throws com.google.protobuf.InvalidProtocolBufferException {\n"
--    "  return newBuilder().mergeFrom(data).buildParsed();\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString data)\n"
-+    "    throws org.apache.pulsar.shaded.com.google.protobuf.v241.InvalidProtocolBufferException {\n"
-+    // "  return newBuilder().mergeFrom(data).buildParsed();\n"
-+    "     throw new RuntimeException(\"Disabled\");\n"
-     "}\n"
-     "public static $classname$ parseFrom(\n"
--    "    com.google.protobuf.ByteString data,\n"
--    "    com.google.protobuf.ExtensionRegistryLite extensionRegistry)\n"
--    "    throws com.google.protobuf.InvalidProtocolBufferException {\n"
--    "  return newBuilder().mergeFrom(data, extensionRegistry)\n"
--    "           .buildParsed();\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString data,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistryLite extensionRegistry)\n"
-+    "    throws org.apache.pulsar.shaded.com.google.protobuf.v241.InvalidProtocolBufferException {\n"
-+    // "  return newBuilder().mergeFrom(data, extensionRegistry)\n"
-+    // "           .buildParsed();\n"
-+    "     throw new RuntimeException(\"Disabled\");\n"
-     "}\n"
-     "public static $classname$ parseFrom(byte[] data)\n"
--    "    throws com.google.protobuf.InvalidProtocolBufferException {\n"
-+    "    throws org.apache.pulsar.shaded.com.google.protobuf.v241.InvalidProtocolBufferException {\n"
-     "  return newBuilder().mergeFrom(data).buildParsed();\n"
-     "}\n"
-     "public static $classname$ parseFrom(\n"
-     "    byte[] data,\n"
--    "    com.google.protobuf.ExtensionRegistryLite extensionRegistry)\n"
--    "    throws com.google.protobuf.InvalidProtocolBufferException {\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistryLite extensionRegistry)\n"
-+    "    throws org.apache.pulsar.shaded.com.google.protobuf.v241.InvalidProtocolBufferException {\n"
-     "  return newBuilder().mergeFrom(data, extensionRegistry)\n"
-     "           .buildParsed();\n"
-     "}\n"
-@@ -593,7 +625,7 @@ GenerateParseFromMethods(io::Printer* printer) {
-     "}\n"
-     "public static $classname$ parseFrom(\n"
-     "    java.io.InputStream input,\n"
--    "    com.google.protobuf.ExtensionRegistryLite extensionRegistry)\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistryLite extensionRegistry)\n"
-     "    throws java.io.IOException {\n"
-     "  return newBuilder().mergeFrom(input, extensionRegistry)\n"
-     "           .buildParsed();\n"
-@@ -609,7 +641,7 @@ GenerateParseFromMethods(io::Printer* printer) {
-     "}\n"
-     "public static $classname$ parseDelimitedFrom(\n"
-     "    java.io.InputStream input,\n"
--    "    com.google.protobuf.ExtensionRegistryLite extensionRegistry)\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistryLite extensionRegistry)\n"
-     "    throws java.io.IOException {\n"
-     "  Builder builder = newBuilder();\n"
-     "  if (builder.mergeDelimitedFrom(input, extensionRegistry)) {\n"
-@@ -619,13 +651,13 @@ GenerateParseFromMethods(io::Printer* printer) {
-     "  }\n"
-     "}\n"
-     "public static $classname$ parseFrom(\n"
--    "    com.google.protobuf.CodedInputStream input)\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.CodedInputStream input)\n"
-     "    throws java.io.IOException {\n"
-     "  return newBuilder().mergeFrom(input).buildParsed();\n"
-     "}\n"
-     "public static $classname$ parseFrom(\n"
--    "    com.google.protobuf.CodedInputStream input,\n"
--    "    com.google.protobuf.ExtensionRegistryLite extensionRegistry)\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.CodedInputStream input,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistryLite extensionRegistry)\n"
-     "    throws java.io.IOException {\n"
-     "  return newBuilder().mergeFrom(input, extensionRegistry)\n"
-     "           .buildParsed();\n"
-@@ -663,7 +695,7 @@ void MessageGenerator::GenerateBuilder(io::Printer* printer) {
-      printer->Print(
-       "@java.lang.Override\n"
-       "protected Builder newBuilderForType(\n"
--      "    com.google.protobuf.GeneratedMessage.BuilderParent parent) {\n"
-+      "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.BuilderParent parent) {\n"
-       "  Builder builder = new Builder(parent);\n"
-       "  return builder;\n"
-       "}\n");
-@@ -673,29 +705,29 @@ void MessageGenerator::GenerateBuilder(io::Printer* printer) {
-     if (HasDescriptorMethods(descriptor_)) {
-       printer->Print(
-         "public static final class Builder extends\n"
--        "    com.google.protobuf.GeneratedMessage.ExtendableBuilder<\n"
--        "      $classname$, Builder> implements $classname$OrBuilder {\n",
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.ExtendableBuilder<\n"
-+        "      $classname$, Builder> implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream.ByteBufMessageBuilder  {\n",
-         "classname", ClassName(descriptor_));
-     } else {
-       printer->Print(
-         "public static final class Builder extends\n"
--        "    com.google.protobuf.GeneratedMessageLite.ExtendableBuilder<\n"
--        "      $classname$, Builder> implements $classname$OrBuilder {\n",
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite.ExtendableBuilder<\n"
-+        "      $classname$, Builder> implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream.ByteBufMessageBuilder  {\n",
-         "classname", ClassName(descriptor_));
-     }
-   } else {
-     if (HasDescriptorMethods(descriptor_)) {
-       printer->Print(
-         "public static final class Builder extends\n"
--        "    com.google.protobuf.GeneratedMessage.Builder<Builder>\n"
--         "   implements $classname$OrBuilder {\n",
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.Builder<Builder>\n"
-+         "   implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream.ByteBufMessageBuilder {\n",
-         "classname", ClassName(descriptor_));
-     } else {
-       printer->Print(
-         "public static final class Builder extends\n"
--        "    com.google.protobuf.GeneratedMessageLite.Builder<\n"
-+        "    org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite.Builder<\n"
-         "      $classname$, Builder>\n"
--        "    implements $classname$OrBuilder {\n",
-+        "    implements $classname$OrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream.ByteBufMessageBuilder  {\n",
-         "classname", ClassName(descriptor_));
-     }
-   }
-@@ -740,12 +772,12 @@ void MessageGenerator::GenerateBuilder(io::Printer* printer) {
- void MessageGenerator::GenerateDescriptorMethods(io::Printer* printer) {
-   if (HasDescriptorMethods(descriptor_)) {
-     printer->Print(
--      "public static final com.google.protobuf.Descriptors.Descriptor\n"
-+      "public static final org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.Descriptor\n"
-       "    getDescriptor() {\n"
-       "  return $fileclass$.internal_$identifier$_descriptor;\n"
-       "}\n"
-       "\n"
--      "protected com.google.protobuf.GeneratedMessage.FieldAccessorTable\n"
-+      "protected org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.FieldAccessorTable\n"
-       "    internalGetFieldAccessorTable() {\n"
-       "  return $fileclass$.internal_$identifier$_fieldAccessorTable;\n"
-       "}\n"
-@@ -760,9 +792,21 @@ void MessageGenerator::GenerateDescriptorMethods(io::Printer* printer) {
- void MessageGenerator::GenerateCommonBuilderMethods(io::Printer* printer) {
-   printer->Print(
-     "// Construct using $classname$.newBuilder()\n"
--    "private Builder() {\n"
-+    "private final io.netty.util.Recycler.Handle handle;\n"
-+    "private Builder(io.netty.util.Recycler.Handle handle) {\n"
-+    "  this.handle = handle;\n"
-     "  maybeForceBuilderInitialization();\n"
-     "}\n"
-+    "private static final io.netty.util.Recycler<Builder> RECYCLER = new io.netty.util.Recycler<Builder>() {\n"
-+    "   protected Builder newObject(io.netty.util.Recycler.Handle handle) {\n"
-+    "         return new Builder(handle);\n"
-+    "       }\n"
-+    "      };\n"
-+    "\n"
-+    " public void recycle() {\n"
-+    "          clear();\n"
-+    "          if (handle != null) {RECYCLER.recycle(this, handle);}\n"
-+    "      }\n"
-     "\n",
-     "classname", ClassName(descriptor_));
- 
-@@ -779,7 +823,7 @@ void MessageGenerator::GenerateCommonBuilderMethods(io::Printer* printer) {
-   if (HasNestedBuilders(descriptor_)) {
-     printer->Print(
-       "private void maybeForceBuilderInitialization() {\n"
--      "  if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) {\n");
-+      "  if (org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.alwaysUseFieldBuilders) {\n");
- 
-     printer->Indent();
-     printer->Indent();
-@@ -801,7 +845,8 @@ void MessageGenerator::GenerateCommonBuilderMethods(io::Printer* printer) {
- 
-   printer->Print(
-     "private static Builder create() {\n"
--    "  return new Builder();\n"
-+    "  return RECYCLER.get();\n"
-+        // "  return new Builder(null);\n"
-     "}\n"
-     "\n"
-     "public Builder clear() {\n"
-@@ -828,7 +873,7 @@ void MessageGenerator::GenerateCommonBuilderMethods(io::Printer* printer) {
-     "classname", ClassName(descriptor_));
-   if (HasDescriptorMethods(descriptor_)) {
-     printer->Print(
--      "public com.google.protobuf.Descriptors.Descriptor\n"
-+      "public org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.Descriptor\n"
-       "    getDescriptorForType() {\n"
-       "  return $classname$.getDescriptor();\n"
-       "}\n"
-@@ -854,7 +899,7 @@ void MessageGenerator::GenerateCommonBuilderMethods(io::Printer* printer) {
-     "}\n"
-     "\n"
-     "private $classname$ buildParsed()\n"
--    "    throws com.google.protobuf.InvalidProtocolBufferException {\n"
-+    "    throws org.apache.pulsar.shaded.com.google.protobuf.v241.InvalidProtocolBufferException {\n"
-     "  $classname$ result = buildPartial();\n"
-     "  if (!result.isInitialized()) {\n"
-     "    throw newUninitializedMessageException(\n"
-@@ -864,7 +909,8 @@ void MessageGenerator::GenerateCommonBuilderMethods(io::Printer* printer) {
-     "}\n"
-     "\n"
-     "public $classname$ buildPartial() {\n"
--    "  $classname$ result = new $classname$(this);\n",
-+    "  $classname$ result = $classname$.RECYCLER.get();\n",
-+        // "  $classname$ result = new $classname$(null);\n",
-     "classname", ClassName(descriptor_));
- 
-   printer->Indent();
-@@ -922,7 +968,7 @@ void MessageGenerator::GenerateCommonBuilderMethods(io::Printer* printer) {
-     // messages type by its descriptor.
-     if (HasDescriptorMethods(descriptor_)) {
-       printer->Print(
--        "public Builder mergeFrom(com.google.protobuf.Message other) {\n"
-+        "public Builder mergeFrom(org.apache.pulsar.shaded.com.google.protobuf.v241.Message other) {\n"
-         "  if (other instanceof $classname$) {\n"
-         "    return mergeFrom(($classname$)other);\n"
-         "  } else {\n"
-@@ -973,16 +1019,21 @@ void MessageGenerator::GenerateBuilderParsingMethods(io::Printer* printer) {
-     SortFieldsByNumber(descriptor_));
- 
-   printer->Print(
-+    "public Builder mergeFrom(org.apache.pulsar.shaded.com.google.protobuf.v241.CodedInputStream input,\n"
-+    "                        org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistryLite extensionRegistry)\n"
-+    "                        throws java.io.IOException {\n"
-+    "   throw new java.io.IOException(\"Merge from CodedInputStream is disabled\");\n"
-+    "                        }\n"
-     "public Builder mergeFrom(\n"
--    "    com.google.protobuf.CodedInputStream input,\n"
--    "    com.google.protobuf.ExtensionRegistryLite extensionRegistry)\n"
-+    "    org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream input,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ExtensionRegistryLite extensionRegistry)\n"
-     "    throws java.io.IOException {\n");
-   printer->Indent();
- 
-   if (HasUnknownFields(descriptor_)) {
-     printer->Print(
--      "com.google.protobuf.UnknownFieldSet.Builder unknownFields =\n"
--      "  com.google.protobuf.UnknownFieldSet.newBuilder(\n"
-+      "org.apache.pulsar.shaded.com.google.protobuf.v241.UnknownFieldSet.Builder unknownFields =\n"
-+      "  org.apache.pulsar.shaded.com.google.protobuf.v241.UnknownFieldSet.newBuilder(\n"
-       "    this.getUnknownFields());\n");
-   }
- 
-@@ -1017,7 +1068,7 @@ void MessageGenerator::GenerateBuilderParsingMethods(io::Printer* printer) {
-       "  $on_changed$\n"
-       "  return this;\n"
-       "default: {\n"
--      "  if (!parseUnknownField(input, extensionRegistry, tag)) {\n"
-+      "  if (!input.skipField(tag)) {\n"
-       "    $on_changed$\n"
-       "    return this;\n"   // it's an endgroup tag
-       "  }\n"
-diff --git a/src/google/protobuf/compiler/java/java_message_field.cc b/src/google/protobuf/compiler/java/java_message_field.cc
-index 251945af..e19cfa58 100644
---- a/src/google/protobuf/compiler/java/java_message_field.cc
-+++ b/src/google/protobuf/compiler/java/java_message_field.cc
-@@ -199,7 +199,7 @@ GenerateBuilderMembers(io::Printer* printer) const {
-     printer->Print(variables_,
-       // If this builder is non-null, it is used and the other fields are
-       // ignored.
--      "private com.google.protobuf.SingleFieldBuilder<\n"
-+      "private org.apache.pulsar.shaded.com.google.protobuf.v241.SingleFieldBuilder<\n"
-       "    $type$, $type$.Builder, $type$OrBuilder> $name$Builder_;"
-       "\n");
-   }
-@@ -295,11 +295,11 @@ GenerateBuilderMembers(io::Printer* printer) const {
-       "    return $name$_;\n"
-       "  }\n"
-       "}\n"
--      "private com.google.protobuf.SingleFieldBuilder<\n"
-+      "private org.apache.pulsar.shaded.com.google.protobuf.v241.SingleFieldBuilder<\n"
-       "    $type$, $type$.Builder, $type$OrBuilder> \n"
-       "    get$capitalized_name$FieldBuilder() {\n"
-       "  if ($name$Builder_ == null) {\n"
--      "    $name$Builder_ = new com.google.protobuf.SingleFieldBuilder<\n"
-+      "    $name$Builder_ = new org.apache.pulsar.shaded.com.google.protobuf.v241.SingleFieldBuilder<\n"
-       "        $type$, $type$.Builder, $type$OrBuilder>(\n"
-       "            $name$_,\n"
-       "            getParentForChildren(),\n"
-@@ -371,7 +371,8 @@ GenerateParsingCode(io::Printer* printer) const {
-   }
- 
-   printer->Print(variables_,
--    "set$capitalized_name$(subBuilder.buildPartial());\n");
-+    "set$capitalized_name$(subBuilder.buildPartial());\n"
-+    "subBuilder.recycle();\n");
- }
- 
- void MessageFieldGenerator::
-@@ -386,7 +387,7 @@ void MessageFieldGenerator::
- GenerateSerializedSizeCode(io::Printer* printer) const {
-   printer->Print(variables_,
-     "if ($get_has_field_bit_message$) {\n"
--    "  size += com.google.protobuf.CodedOutputStream\n"
-+    "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-     "    .compute$group_or_message$Size($number$, $name$_);\n"
-     "}\n");
- }
-@@ -543,7 +544,7 @@ GenerateBuilderMembers(io::Printer* printer) const {
-     printer->Print(variables_,
-       // If this builder is non-null, it is used and the other fields are
-       // ignored.
--      "private com.google.protobuf.RepeatedFieldBuilder<\n"
-+      "private org.apache.pulsar.shaded.com.google.protobuf.v241.RepeatedFieldBuilder<\n"
-       "    $type$, $type$.Builder, $type$OrBuilder> $name$Builder_;\n"
-       "\n");
-   }
-@@ -738,11 +739,11 @@ GenerateBuilderMembers(io::Printer* printer) const {
-       "     get$capitalized_name$BuilderList() {\n"
-       "  return get$capitalized_name$FieldBuilder().getBuilderList();\n"
-       "}\n"
--      "private com.google.protobuf.RepeatedFieldBuilder<\n"
-+      "private org.apache.pulsar.shaded.com.google.protobuf.v241.RepeatedFieldBuilder<\n"
-       "    $type$, $type$.Builder, $type$OrBuilder> \n"
-       "    get$capitalized_name$FieldBuilder() {\n"
-       "  if ($name$Builder_ == null) {\n"
--      "    $name$Builder_ = new com.google.protobuf.RepeatedFieldBuilder<\n"
-+      "    $name$Builder_ = new org.apache.pulsar.shaded.com.google.protobuf.v241.RepeatedFieldBuilder<\n"
-       "        $type$, $type$.Builder, $type$OrBuilder>(\n"
-       "            $name$_,\n"
-       "            $get_mutable_bit_builder$,\n"
-@@ -801,7 +802,7 @@ GenerateMergingCode(io::Printer* printer) const {
-     "    $name$_ = other.$name$_;\n"
-     "    $clear_mutable_bit_builder$;\n"
-     "    $name$Builder_ = \n"
--    "      com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders ?\n"
-+    "      org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessage.alwaysUseFieldBuilders ?\n"
-     "         get$capitalized_name$FieldBuilder() : null;\n"
-     "  } else {\n"
-     "    $name$Builder_.addAllMessages(other.$name$_);\n"
-@@ -853,7 +854,7 @@ void RepeatedMessageFieldGenerator::
- GenerateSerializedSizeCode(io::Printer* printer) const {
-   printer->Print(variables_,
-     "for (int i = 0; i < $name$_.size(); i++) {\n"
--    "  size += com.google.protobuf.CodedOutputStream\n"
-+    "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-     "    .compute$group_or_message$Size($number$, $name$_.get(i));\n"
-     "}\n");
- }
-diff --git a/src/google/protobuf/compiler/java/java_primitive_field.cc b/src/google/protobuf/compiler/java/java_primitive_field.cc
-index addb8819..3d72a15d 100644
---- a/src/google/protobuf/compiler/java/java_primitive_field.cc
-+++ b/src/google/protobuf/compiler/java/java_primitive_field.cc
-@@ -60,7 +60,7 @@ const char* PrimitiveTypeName(JavaType type) {
-     case JAVATYPE_DOUBLE : return "double";
-     case JAVATYPE_BOOLEAN: return "boolean";
-     case JAVATYPE_STRING : return "java.lang.String";
--    case JAVATYPE_BYTES  : return "com.google.protobuf.ByteString";
-+    case JAVATYPE_BYTES  : return "org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString";
-     case JAVATYPE_ENUM   : return NULL;
-     case JAVATYPE_MESSAGE: return NULL;
- 
-@@ -351,7 +351,7 @@ void PrimitiveFieldGenerator::
- GenerateSerializedSizeCode(io::Printer* printer) const {
-   printer->Print(variables_,
-     "if ($get_has_field_bit_message$) {\n"
--    "  size += com.google.protobuf.CodedOutputStream\n"
-+    "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-     "    .compute$capitalized_type$Size($number$, $name$_);\n"
-     "}\n");
- }
-@@ -660,7 +660,7 @@ GenerateSerializedSizeCode(io::Printer* printer) const {
-   if (FixedSize(GetType(descriptor_)) == -1) {
-     printer->Print(variables_,
-       "for (int i = 0; i < $name$_.size(); i++) {\n"
--      "  dataSize += com.google.protobuf.CodedOutputStream\n"
-+      "  dataSize += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-       "    .compute$capitalized_type$SizeNoTag($name$_.get(i));\n"
-       "}\n");
-   } else {
-@@ -675,7 +675,7 @@ GenerateSerializedSizeCode(io::Printer* printer) const {
-     printer->Print(variables_,
-       "if (!get$capitalized_name$List().isEmpty()) {\n"
-       "  size += $tag_size$;\n"
--      "  size += com.google.protobuf.CodedOutputStream\n"
-+      "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-       "      .computeInt32SizeNoTag(dataSize);\n"
-       "}\n");
-   } else {
-diff --git a/src/google/protobuf/compiler/java/java_service.cc b/src/google/protobuf/compiler/java/java_service.cc
-index 1ae4f461..831aec03 100644
---- a/src/google/protobuf/compiler/java/java_service.cc
-+++ b/src/google/protobuf/compiler/java/java_service.cc
-@@ -52,7 +52,7 @@ void ServiceGenerator::Generate(io::Printer* printer) {
-   bool is_own_file = descriptor_->file()->options().java_multiple_files();
-   printer->Print(
-     "public $static$ abstract class $classname$\n"
--    "    implements com.google.protobuf.Service {\n",
-+    "    implements org.apache.pulsar.shaded.com.google.protobuf.v241.Service {\n",
-     "static", is_own_file ? "" : "static",
-     "classname", descriptor_->name());
-   printer->Indent();
-@@ -71,7 +71,7 @@ void ServiceGenerator::Generate(io::Printer* printer) {
-   // Generate getDescriptor() and getDescriptorForType().
-   printer->Print(
-     "public static final\n"
--    "    com.google.protobuf.Descriptors.ServiceDescriptor\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.ServiceDescriptor\n"
-     "    getDescriptor() {\n"
-     "  return $file$.getDescriptor().getServices().get($index$);\n"
-     "}\n",
-@@ -92,7 +92,7 @@ void ServiceGenerator::Generate(io::Printer* printer) {
- 
- void ServiceGenerator::GenerateGetDescriptorForType(io::Printer* printer) {
-   printer->Print(
--    "public final com.google.protobuf.Descriptors.ServiceDescriptor\n"
-+    "public final org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.ServiceDescriptor\n"
-     "    getDescriptorForType() {\n"
-     "  return getDescriptor();\n"
-     "}\n");
-@@ -109,7 +109,7 @@ void ServiceGenerator::GenerateInterface(io::Printer* printer) {
- void ServiceGenerator::GenerateNewReflectiveServiceMethod(
-     io::Printer* printer) {
-   printer->Print(
--    "public static com.google.protobuf.Service newReflectiveService(\n"
-+    "public static org.apache.pulsar.shaded.com.google.protobuf.v241.Service newReflectiveService(\n"
-     "    final Interface impl) {\n"
-     "  return new $classname$() {\n",
-     "classname", descriptor_->name());
-@@ -136,9 +136,9 @@ void ServiceGenerator::GenerateNewReflectiveServiceMethod(
- void ServiceGenerator::GenerateNewReflectiveBlockingServiceMethod(
-     io::Printer* printer) {
-   printer->Print(
--    "public static com.google.protobuf.BlockingService\n"
-+    "public static org.apache.pulsar.shaded.com.google.protobuf.v241.BlockingService\n"
-     "    newReflectiveBlockingService(final BlockingInterface impl) {\n"
--    "  return new com.google.protobuf.BlockingService() {\n");
-+    "  return new org.apache.pulsar.shaded.com.google.protobuf.v241.BlockingService() {\n");
-   printer->Indent();
-   printer->Indent();
- 
-@@ -166,11 +166,11 @@ void ServiceGenerator::GenerateCallMethod(io::Printer* printer) {
-   printer->Print(
-     "\n"
-     "public final void callMethod(\n"
--    "    com.google.protobuf.Descriptors.MethodDescriptor method,\n"
--    "    com.google.protobuf.RpcController controller,\n"
--    "    com.google.protobuf.Message request,\n"
--    "    com.google.protobuf.RpcCallback<\n"
--    "      com.google.protobuf.Message> done) {\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.MethodDescriptor method,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcController controller,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.Message request,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcCallback<\n"
-+    "      org.apache.pulsar.shaded.com.google.protobuf.v241.Message> done) {\n"
-     "  if (method.getService() != getDescriptor()) {\n"
-     "    throw new java.lang.IllegalArgumentException(\n"
-     "      \"Service.callMethod() given method descriptor for wrong \" +\n"
-@@ -190,7 +190,7 @@ void ServiceGenerator::GenerateCallMethod(io::Printer* printer) {
-     printer->Print(vars,
-       "case $index$:\n"
-       "  this.$method$(controller, ($input$)request,\n"
--      "    com.google.protobuf.RpcUtil.<$output$>specializeCallback(\n"
-+      "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcUtil.<$output$>specializeCallback(\n"
-       "      done));\n"
-       "  return;\n");
-   }
-@@ -211,11 +211,11 @@ void ServiceGenerator::GenerateCallMethod(io::Printer* printer) {
- void ServiceGenerator::GenerateCallBlockingMethod(io::Printer* printer) {
-   printer->Print(
-     "\n"
--    "public final com.google.protobuf.Message callBlockingMethod(\n"
--    "    com.google.protobuf.Descriptors.MethodDescriptor method,\n"
--    "    com.google.protobuf.RpcController controller,\n"
--    "    com.google.protobuf.Message request)\n"
--    "    throws com.google.protobuf.ServiceException {\n"
-+    "public final org.apache.pulsar.shaded.com.google.protobuf.v241.Message callBlockingMethod(\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.MethodDescriptor method,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcController controller,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.Message request)\n"
-+    "    throws org.apache.pulsar.shaded.com.google.protobuf.v241.ServiceException {\n"
-     "  if (method.getService() != getDescriptor()) {\n"
-     "    throw new java.lang.IllegalArgumentException(\n"
-     "      \"Service.callBlockingMethod() given method descriptor for \" +\n"
-@@ -257,9 +257,9 @@ void ServiceGenerator::GenerateGetPrototype(RequestOrResponse which,
-    * "BlockingService.foo."  Consider fixing.
-    */
-   printer->Print(
--    "public final com.google.protobuf.Message\n"
-+    "public final org.apache.pulsar.shaded.com.google.protobuf.v241.Message\n"
-     "    get$request_or_response$Prototype(\n"
--    "    com.google.protobuf.Descriptors.MethodDescriptor method) {\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.Descriptors.MethodDescriptor method) {\n"
-     "  if (method.getService() != getDescriptor()) {\n"
-     "    throw new java.lang.IllegalArgumentException(\n"
-     "      \"Service.get$request_or_response$Prototype() given method \" +\n"
-@@ -297,7 +297,7 @@ void ServiceGenerator::GenerateGetPrototype(RequestOrResponse which,
- void ServiceGenerator::GenerateStub(io::Printer* printer) {
-   printer->Print(
-     "public static Stub newStub(\n"
--    "    com.google.protobuf.RpcChannel channel) {\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcChannel channel) {\n"
-     "  return new Stub(channel);\n"
-     "}\n"
-     "\n"
-@@ -307,13 +307,13 @@ void ServiceGenerator::GenerateStub(io::Printer* printer) {
-   printer->Indent();
- 
-   printer->Print(
--    "private Stub(com.google.protobuf.RpcChannel channel) {\n"
-+    "private Stub(org.apache.pulsar.shaded.com.google.protobuf.v241.RpcChannel channel) {\n"
-     "  this.channel = channel;\n"
-     "}\n"
-     "\n"
--    "private final com.google.protobuf.RpcChannel channel;\n"
-+    "private final org.apache.pulsar.shaded.com.google.protobuf.v241.RpcChannel channel;\n"
-     "\n"
--    "public com.google.protobuf.RpcChannel getChannel() {\n"
-+    "public org.apache.pulsar.shaded.com.google.protobuf.v241.RpcChannel getChannel() {\n"
-     "  return channel;\n"
-     "}\n");
- 
-@@ -333,7 +333,7 @@ void ServiceGenerator::GenerateStub(io::Printer* printer) {
-       "  controller,\n"
-       "  request,\n"
-       "  $output$.getDefaultInstance(),\n"
--      "  com.google.protobuf.RpcUtil.generalizeCallback(\n"
-+      "  org.apache.pulsar.shaded.com.google.protobuf.v241.RpcUtil.generalizeCallback(\n"
-       "    done,\n"
-       "    $output$.class,\n"
-       "    $output$.getDefaultInstance()));\n");
-@@ -351,7 +351,7 @@ void ServiceGenerator::GenerateStub(io::Printer* printer) {
- void ServiceGenerator::GenerateBlockingStub(io::Printer* printer) {
-   printer->Print(
-     "public static BlockingInterface newBlockingStub(\n"
--    "    com.google.protobuf.BlockingRpcChannel channel) {\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.BlockingRpcChannel channel) {\n"
-     "  return new BlockingStub(channel);\n"
-     "}\n"
-     "\n");
-@@ -376,11 +376,11 @@ void ServiceGenerator::GenerateBlockingStub(io::Printer* printer) {
-   printer->Indent();
- 
-   printer->Print(
--    "private BlockingStub(com.google.protobuf.BlockingRpcChannel channel) {\n"
-+    "private BlockingStub(org.apache.pulsar.shaded.com.google.protobuf.v241.BlockingRpcChannel channel) {\n"
-     "  this.channel = channel;\n"
-     "}\n"
-     "\n"
--    "private final com.google.protobuf.BlockingRpcChannel channel;\n");
-+    "private final org.apache.pulsar.shaded.com.google.protobuf.v241.BlockingRpcChannel channel;\n");
- 
-   for (int i = 0; i < descriptor_->method_count(); i++) {
-     const MethodDescriptor* method = descriptor_->method(i);
-@@ -418,9 +418,9 @@ void ServiceGenerator::GenerateMethodSignature(io::Printer* printer,
-   vars["abstract"] = (is_abstract == IS_ABSTRACT) ? "abstract" : "";
-   printer->Print(vars,
-     "public $abstract$ void $name$(\n"
--    "    com.google.protobuf.RpcController controller,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcController controller,\n"
-     "    $input$ request,\n"
--    "    com.google.protobuf.RpcCallback<$output$> done)");
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcCallback<$output$> done)");
- }
- 
- void ServiceGenerator::GenerateBlockingMethodSignature(
-@@ -433,9 +433,9 @@ void ServiceGenerator::GenerateBlockingMethodSignature(
-   printer->Print(vars,
-     "\n"
-     "public $output$ $method$(\n"
--    "    com.google.protobuf.RpcController controller,\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.RpcController controller,\n"
-     "    $input$ request)\n"
--    "    throws com.google.protobuf.ServiceException");
-+    "    throws org.apache.pulsar.shaded.com.google.protobuf.v241.ServiceException");
- }
- 
- }  // namespace java
-diff --git a/src/google/protobuf/compiler/java/java_string_field.cc b/src/google/protobuf/compiler/java/java_string_field.cc
-index 4f1fb14c..f52a1120 100644
---- a/src/google/protobuf/compiler/java/java_string_field.cc
-+++ b/src/google/protobuf/compiler/java/java_string_field.cc
-@@ -63,7 +63,7 @@ void SetPrimitiveVariables(const FieldDescriptor* descriptor,
-     UnderscoresToCapitalizedCamelCase(descriptor);
-   (*variables)["constant_name"] = FieldConstantName(descriptor);
-   (*variables)["number"] = SimpleItoa(descriptor->number());
--  (*variables)["empty_list"] = "com.google.protobuf.LazyStringArrayList.EMPTY";
-+  (*variables)["empty_list"] = "org.apache.pulsar.shaded.com.google.protobuf.v241.LazyStringArrayList.EMPTY";
- 
-   (*variables)["default"] = DefaultValue(descriptor);
-   (*variables)["default_init"] = ("= " + DefaultValue(descriptor));
-@@ -179,24 +179,24 @@ GenerateMembers(io::Printer* printer) const {
-     "  if (ref instanceof String) {\n"
-     "    return (String) ref;\n"
-     "  } else {\n"
--    "    com.google.protobuf.ByteString bs = \n"
--    "        (com.google.protobuf.ByteString) ref;\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString bs = \n"
-+    "        (org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString) ref;\n"
-     "    String s = bs.toStringUtf8();\n"
--    "    if (com.google.protobuf.Internal.isValidUtf8(bs)) {\n"
-+    "    if (org.apache.pulsar.shaded.com.google.protobuf.v241.Internal.isValidUtf8(bs)) {\n"
-     "      $name$_ = s;\n"
-     "    }\n"
-     "    return s;\n"
-     "  }\n"
-     "}\n"
--    "private com.google.protobuf.ByteString get$capitalized_name$Bytes() {\n"
-+    "private org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString get$capitalized_name$Bytes() {\n"
-     "  java.lang.Object ref = $name$_;\n"
-     "  if (ref instanceof String) {\n"
--    "    com.google.protobuf.ByteString b = \n"
--    "        com.google.protobuf.ByteString.copyFromUtf8((String) ref);\n"
-+    "    org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString b = \n"
-+    "        org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString.copyFromUtf8((String) ref);\n"
-     "    $name$_ = b;\n"
-     "    return b;\n"
-     "  } else {\n"
--    "    return (com.google.protobuf.ByteString) ref;\n"
-+    "    return (org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString) ref;\n"
-     "  }\n"
-     "}\n");
- }
-@@ -213,7 +213,7 @@ GenerateBuilderMembers(io::Printer* printer) const {
-     "$deprecation$public String get$capitalized_name$() {\n"
-     "  java.lang.Object ref = $name$_;\n"
-     "  if (!(ref instanceof String)) {\n"
--    "    String s = ((com.google.protobuf.ByteString) ref).toStringUtf8();\n"
-+    "    String s = ((org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString) ref).toStringUtf8();\n"
-     "    $name$_ = s;\n"
-     "    return s;\n"
-     "  } else {\n"
-@@ -241,7 +241,7 @@ GenerateBuilderMembers(io::Printer* printer) const {
-     "}\n");
- 
-   printer->Print(variables_,
--    "void set$capitalized_name$(com.google.protobuf.ByteString value) {\n"
-+    "void set$capitalized_name$(org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString value) {\n"
-     "  $set_has_field_bit_builder$;\n"
-     "  $name$_ = value;\n"
-     "  $on_changed$\n"
-@@ -301,7 +301,7 @@ void StringFieldGenerator::
- GenerateSerializedSizeCode(io::Printer* printer) const {
-   printer->Print(variables_,
-     "if ($get_has_field_bit_message$) {\n"
--    "  size += com.google.protobuf.CodedOutputStream\n"
-+    "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-     "    .computeBytesSize($number$, get$capitalized_name$Bytes());\n"
-     "}\n");
- }
-@@ -360,7 +360,7 @@ GenerateInterfaceMembers(io::Printer* printer) const {
- void RepeatedStringFieldGenerator::
- GenerateMembers(io::Printer* printer) const {
-   printer->Print(variables_,
--    "private com.google.protobuf.LazyStringList $name$_;\n"
-+    "private org.apache.pulsar.shaded.com.google.protobuf.v241.LazyStringList $name$_;\n"
-     "$deprecation$public java.util.List<String>\n"
-     "    get$capitalized_name$List() {\n"
-     "  return $name$_;\n"   // note:  unmodifiable list
-@@ -391,12 +391,12 @@ GenerateBuilderMembers(io::Printer* printer) const {
-   // just that the list cannot be modified via the reference but that the
-   // list can never be modified.
-   printer->Print(variables_,
--    "private com.google.protobuf.LazyStringList $name$_ = $empty_list$;\n");
-+    "private org.apache.pulsar.shaded.com.google.protobuf.v241.LazyStringList $name$_ = $empty_list$;\n");
- 
-   printer->Print(variables_,
-     "private void ensure$capitalized_name$IsMutable() {\n"
-     "  if (!$get_mutable_bit_builder$) {\n"
--    "    $name$_ = new com.google.protobuf.LazyStringArrayList($name$_);\n"
-+    "    $name$_ = new org.apache.pulsar.shaded.com.google.protobuf.v241.LazyStringArrayList($name$_);\n"
-     "    $set_mutable_bit_builder$;\n"
-     "   }\n"
-     "}\n");
-@@ -446,7 +446,7 @@ GenerateBuilderMembers(io::Printer* printer) const {
-     "}\n");
- 
-   printer->Print(variables_,
--    "void add$capitalized_name$(com.google.protobuf.ByteString value) {\n"
-+    "void add$capitalized_name$(org.apache.pulsar.shaded.com.google.protobuf.v241.ByteString value) {\n"
-     "  ensure$capitalized_name$IsMutable();\n"
-     "  $name$_.add(value);\n"
-     "  $on_changed$\n"
-@@ -497,7 +497,7 @@ GenerateBuildingCode(io::Printer* printer) const {
- 
-   printer->Print(variables_,
-     "if ($get_mutable_bit_builder$) {\n"
--    "  $name$_ = new com.google.protobuf.UnmodifiableLazyStringList(\n"
-+    "  $name$_ = new org.apache.pulsar.shaded.com.google.protobuf.v241.UnmodifiableLazyStringList(\n"
-     "      $name$_);\n"
-     "  $clear_mutable_bit_builder$;\n"
-     "}\n"
-@@ -550,7 +550,7 @@ GenerateSerializedSizeCode(io::Printer* printer) const {
- 
-   printer->Print(variables_,
-     "for (int i = 0; i < $name$_.size(); i++) {\n"
--    "  dataSize += com.google.protobuf.CodedOutputStream\n"
-+    "  dataSize += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-     "    .computeBytesSizeNoTag($name$_.getByteString(i));\n"
-     "}\n");
- 
-@@ -561,7 +561,7 @@ GenerateSerializedSizeCode(io::Printer* printer) const {
-     printer->Print(variables_,
-       "if (!get$capitalized_name$List().isEmpty()) {\n"
-       "  size += $tag_size$;\n"
--      "  size += com.google.protobuf.CodedOutputStream\n"
-+      "  size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream\n"
-       "      .computeInt32SizeNoTag(dataSize);\n"
-       "}\n");
-   } else {
-diff --git a/src/google/protobuf/message.cc b/src/google/protobuf/message.cc
-index 91e6878e..cfee3832 100644
---- a/src/google/protobuf/message.cc
-+++ b/src/google/protobuf/message.cc
-@@ -32,6 +32,7 @@
- //  Based on original Protocol Buffers design by
- //  Sanjay Ghemawat, Jeff Dean, and others.
- 
-+#include <iostream>
- #include <stack>
- #include <google/protobuf/stubs/hash.h>
- 
diff --git a/pulsar-client-cpp/CMakeLists.txt b/pulsar-client-cpp/CMakeLists.txt
index 5cbbea6780b52..ca98dac9b48a0 100644
--- a/pulsar-client-cpp/CMakeLists.txt
+++ b/pulsar-client-cpp/CMakeLists.txt
@@ -123,12 +123,12 @@ unset(OPENSSL_VERSION CACHE)
 
 if (LINK_STATIC)
     find_library(ZLIB_LIBRARIES REQUIRED NAMES libz.a z zlib)
-    find_library(Protobuf_LITE_LIBRARIES NAMES libprotobuf-lite.a libprotobuf-lite)
+    find_library(Protobuf_LIBRARIES NAMES libprotobuf.a libprotobuf)
     find_library(CURL_LIBRARIES NAMES libcurl.a curl curl_a libcurl_a)
     find_library(LIB_ZSTD NAMES libzstd.a)
     find_library(LIB_SNAPPY NAMES libsnappy.a)
-    message(STATUS "Protobuf_LITE_LIBRARIES: ${Protobuf_LITE_LIBRARIES}")
-    set(COMMON_LIBS ${Protobuf_LITE_LIBRARIES} ${COMMON_LIBS})
+    message(STATUS "Protobuf_LIBRARIES: ${Protobuf_LIBRARIES}")
+    set(COMMON_LIBS ${Protobuf_LIBRARIES} ${COMMON_LIBS})
 
     if (USE_LOG4CXX)
         if (LOG4CXX_USE_DYNAMIC_LIBS)
@@ -163,28 +163,30 @@ else()
     find_package(ZLIB REQUIRED)
     set(ZLIB_LIBRARIES ${ZLIB_LIBRARIES})
     # NOTE: The default MODULE mode may not find debug libraries so use CONFIG mode here
+    unset(Protobuf_INCLUDE_DIRS CACHE)
+    unset(Protobuf_LIBRARIES CACHE)
     find_package(Protobuf QUIET CONFIG)
     # NOTE: On Windows x86 platform, Protobuf_FOUND might be set false but Protobuf_INCLUDE_DIRS and
-    # Protobuf_LITE_LIBRARIES are both found.
-    if (Protobuf_INCLUDE_DIRS AND Protobuf_LITE_LIBRARIES AND NOT Protobuf_FOUND)
+    # Protobuf_LIBRARIES are both found.
+    if (Protobuf_INCLUDE_DIRS AND Protobuf_LIBRARIES AND NOT Protobuf_FOUND)
         set(Protobuf_FOUND TRUE)
     endif ()
     if (Protobuf_FOUND)
         message("Found Protobuf in config mode")
-        message(STATUS "Protobuf_LITE_LIBRARIES: ${Protobuf_LITE_LIBRARIES}")
+        message(STATUS "Protobuf_LIBRARIES: ${Protobuf_LIBRARIES}")
         message(STATUS "Protobuf_INCLUDE_DIRS: ${Protobuf_INCLUDE_DIRS}")
     else ()
         message("Failed to find Protobuf in config mode, try to find it from system path")
-        find_library(Protobuf_LITE_LIBRARIES protobuf-lite libprotobuf-lite)
+        find_library(Protobuf_LIBRARIES protobuf libprotobuf)
         find_path(Protobuf_INCLUDE_DIRS google/protobuf/stubs/common.h)
-        message(STATUS "Protobuf_LITE_LIBRARIES: ${Protobuf_LITE_LIBRARIES}")
+        message(STATUS "Protobuf_LIBRARIES: ${Protobuf_LIBRARIES}")
         message(STATUS "Protobuf_INCLUDE_DIRS: ${Protobuf_INCLUDE_DIRS}")
     endif ()
 
     if (${Protobuf_FOUND} AND (${CMAKE_VERSION} VERSION_GREATER 3.8))
-        set(COMMON_LIBS protobuf::libprotobuf-lite ${COMMON_LIBS})
+        set(COMMON_LIBS protobuf::libprotobuf ${COMMON_LIBS})
     else ()
-        set(COMMON_LIBS ${Protobuf_LITE_LIBRARIES} ${COMMON_LIBS})
+        set(COMMON_LIBS ${Protobuf_LIBRARIES} ${COMMON_LIBS})
     endif ()
 
     if (MSVC AND (${CMAKE_BUILD_TYPE} STREQUAL Debug))
diff --git a/pulsar-client-cpp/README.md b/pulsar-client-cpp/README.md
index 49c33337678c8..6316df0ce5d51 100644
--- a/pulsar-client-cpp/README.md
+++ b/pulsar-client-cpp/README.md
@@ -40,7 +40,7 @@ https://github.com/apache/pulsar/tree/master/pulsar-client-cpp/examples
  * A C++ compiler that supports C++11, like GCC >= 4.8
  * CMake >= 3.4
  * [Boost](http://www.boost.org/)
- * [Protocol Buffer](https://developers.google.com/protocol-buffers/)
+ * [Protocol Buffer](https://developers.google.com/protocol-buffers/) >= 3
  * [libcurl](https://curl.se/libcurl/)
  * [openssl](https://github.com/openssl/openssl)
 
diff --git a/pulsar-client-cpp/docker-build.sh b/pulsar-client-cpp/docker-build.sh
index b615ae595985f..f3b0a5f98dba2 100755
--- a/pulsar-client-cpp/docker-build.sh
+++ b/pulsar-client-cpp/docker-build.sh
@@ -27,7 +27,7 @@ ROOT_DIR=$(git rev-parse --show-toplevel)
 cd $ROOT_DIR/pulsar-client-cpp
 
 BUILD_IMAGE_NAME="${BUILD_IMAGE_NAME:-apachepulsar/pulsar-build}"
-BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-16.04-py2}"
+BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-16.04-pb3}"
 
 IMAGE="$BUILD_IMAGE_NAME:$BUILD_IMAGE_VERSION"
 
diff --git a/pulsar-client-cpp/docker-format.sh b/pulsar-client-cpp/docker-format.sh
index 53ee8cb675de9..b11f09cad04e9 100755
--- a/pulsar-client-cpp/docker-format.sh
+++ b/pulsar-client-cpp/docker-format.sh
@@ -27,7 +27,7 @@ ROOT_DIR=$(git rev-parse --show-toplevel)
 cd $ROOT_DIR/pulsar-client-cpp
 
 BUILD_IMAGE_NAME="${BUILD_IMAGE_NAME:-apachepulsar/pulsar-build}"
-BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-16.04}"
+BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-16.04-pb3}"
 
 IMAGE="$BUILD_IMAGE_NAME:$BUILD_IMAGE_VERSION"
 
diff --git a/pulsar-client-cpp/docker-tests.sh b/pulsar-client-cpp/docker-tests.sh
index 79a6ded9b2937..bb6565916dd8f 100755
--- a/pulsar-client-cpp/docker-tests.sh
+++ b/pulsar-client-cpp/docker-tests.sh
@@ -34,7 +34,7 @@ ROOT_DIR=$(git rev-parse --show-toplevel)
 cd $ROOT_DIR/pulsar-client-cpp
 
 BUILD_IMAGE_NAME="${BUILD_IMAGE_NAME:-apachepulsar/pulsar-build}"
-BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-16.04-py2}"
+BUILD_IMAGE_VERSION="${BUILD_IMAGE_VERSION:-ubuntu-16.04-pb3}"
 
 IMAGE="$BUILD_IMAGE_NAME:$BUILD_IMAGE_VERSION"
 
@@ -82,4 +82,4 @@ if [ $RES -ne 0 ]; then
   fi
   )
 fi
-exit $RES
\ No newline at end of file
+exit $RES
diff --git a/pulsar-client-cpp/docker/centos-7/Dockerfile b/pulsar-client-cpp/docker/centos-7/Dockerfile
index fb157e1d0f661..53ce9bdbba4f8 100644
--- a/pulsar-client-cpp/docker/centos-7/Dockerfile
+++ b/pulsar-client-cpp/docker/centos-7/Dockerfile
@@ -21,10 +21,16 @@
 FROM centos:7.6.1810
 
 RUN yum install -y gcc gcc-c++ make \
-  protobuf-devel.x86_64 protobuf-lite-devel.x86_64 \
+  protobuf-devel.x86_64 \
   libcurl-devel openssl-devel \
   boost boost-devel
 
+RUN curl -O -L https://github.com/protocolbuffers/protobuf/releases/download/v3.17.3/protobuf-cpp-3.17.3.tar.gz && \
+    tar xfz protobuf-cpp-3.17.3.tar.gz && \
+    cd protobuf-3.17.3/ && \
+    CXXFLAGS=-fPIC ./configure && \
+    make -j8 && make install && \
+    cd .. && rm -rf protobuf-3.17.3/ protobuf-cpp-3.17.3.tar.gz
 RUN mkdir -p /opt/cmake
 WORKDIR /opt/cmake
 RUN curl -L -O https://cmake.org/files/v3.4/cmake-3.4.0-Linux-x86_64.tar.gz \
diff --git a/pulsar-client-cpp/include/pulsar/ProtobufNativeSchema.h b/pulsar-client-cpp/include/pulsar/ProtobufNativeSchema.h
new file mode 100644
index 0000000000000..ef9a7b1c0485c
--- /dev/null
+++ b/pulsar-client-cpp/include/pulsar/ProtobufNativeSchema.h
@@ -0,0 +1,35 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include <pulsar/Schema.h>
+#include <google/protobuf/descriptor.h>
+
+namespace pulsar {
+
+/**
+ * Create a protobuf native schema using a descriptor.
+ *
+ * @param descriptor the Descriptor object of the target class
+ * @return the protobuf native schema
+ * @throw std::invalid_argument if descriptor is nullptr
+ */
+PULSAR_PUBLIC SchemaInfo createProtobufNativeSchema(const google::protobuf::Descriptor* descriptor);
+
+}  // namespace pulsar
diff --git a/pulsar-client-cpp/include/pulsar/Schema.h b/pulsar-client-cpp/include/pulsar/Schema.h
index d8b47f5701652..7e7a5aedbbb63 100644
--- a/pulsar-client-cpp/include/pulsar/Schema.h
+++ b/pulsar-client-cpp/include/pulsar/Schema.h
@@ -89,6 +89,11 @@ enum SchemaType
      */
     KEY_VALUE = 15,
 
+    /**
+     * Protobuf native schema based on Descriptor.
+     */
+    PROTOBUF_NATIVE = 20,
+
     /**
      * A bytes array.
      */
diff --git a/pulsar-client-cpp/lib/Commands.cc b/pulsar-client-cpp/lib/Commands.cc
index 5b17d24c0fc6f..0df18a9a7a471 100644
--- a/pulsar-client-cpp/lib/Commands.cc
+++ b/pulsar-client-cpp/lib/Commands.cc
@@ -42,6 +42,7 @@ static inline bool isBuiltInSchema(SchemaType schemaType) {
         case JSON:
         case AVRO:
         case PROTOBUF:
+        case PROTOBUF_NATIVE:
             return true;
 
         default:
@@ -61,6 +62,8 @@ static inline proto::Schema_Type getSchemaType(SchemaType type) {
             return Schema_Type_Protobuf;
         case AVRO:
             return Schema_Type_Avro;
+        case PROTOBUF_NATIVE:
+            return Schema_Type_ProtobufNative;
         default:
             return Schema_Type_None;
     }
diff --git a/pulsar-client-cpp/lib/ProtobufNativeSchema.cc b/pulsar-client-cpp/lib/ProtobufNativeSchema.cc
new file mode 100644
index 0000000000000..d11f029ae2880
--- /dev/null
+++ b/pulsar-client-cpp/lib/ProtobufNativeSchema.cc
@@ -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
+ *
+ *   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.
+ */
+#include "pulsar/ProtobufNativeSchema.h"
+
+#include <stdexcept>
+#include <vector>
+
+#include <boost/archive/iterators/base64_from_binary.hpp>
+#include <boost/archive/iterators/transform_width.hpp>
+#include <google/protobuf/descriptor.pb.h>
+
+using google::protobuf::FileDescriptor;
+using google::protobuf::FileDescriptorSet;
+
+namespace pulsar {
+
+void internalCollectFileDescriptors(const FileDescriptor* fileDescriptor,
+                                    FileDescriptorSet& fileDescriptorSet);
+
+SchemaInfo createProtobufNativeSchema(const google::protobuf::Descriptor* descriptor) {
+    if (!descriptor) {
+        throw std::invalid_argument("descriptor is null");
+    }
+
+    const auto fileDescriptor = descriptor->file();
+    const std::string rootMessageTypeName = descriptor->full_name();
+    const std::string rootFileDescriptorName = fileDescriptor->name();
+
+    FileDescriptorSet fileDescriptorSet;
+    internalCollectFileDescriptors(fileDescriptor, fileDescriptorSet);
+
+    using namespace boost::archive::iterators;
+    using base64 = base64_from_binary<transform_width<const char*, 6, 8>>;
+
+    std::vector<char> bytes(fileDescriptorSet.ByteSizeLong());
+    fileDescriptorSet.SerializeToArray(bytes.data(), bytes.size());
+
+    const std::string schemaJson =
+        R"({"fileDescriptorSet":")" + std::string(base64(bytes.data()), base64(bytes.data() + bytes.size())) +
+        R"(","rootMessageTypeName":")" + rootMessageTypeName +
+        R"(","rootFileDescriptorName":")" + rootFileDescriptorName + R"("})";
+
+    return SchemaInfo(SchemaType::PROTOBUF_NATIVE, "", schemaJson);
+}
+
+void internalCollectFileDescriptors(const FileDescriptor* fileDescriptor,
+                                    FileDescriptorSet& fileDescriptorSet) {
+    fileDescriptor->CopyTo(fileDescriptorSet.add_file());
+    for (int i = 0; i < fileDescriptor->dependency_count(); i++) {
+        // collect the file descriptors recursively
+        internalCollectFileDescriptors(fileDescriptor->dependency(i), fileDescriptorSet);
+    }
+}
+
+}  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/Schema.cc b/pulsar-client-cpp/lib/Schema.cc
index f883540eedcd3..1f70406d13d6e 100644
--- a/pulsar-client-cpp/lib/Schema.cc
+++ b/pulsar-client-cpp/lib/Schema.cc
@@ -61,6 +61,8 @@ PULSAR_PUBLIC const char *strSchemaType(SchemaType schemaType) {
             return "AUTO_PUBLISH";
         case KEY_VALUE:
             return "KEY_VALUE";
+        case PROTOBUF_NATIVE:
+            return "PROTOBUF_NATIVE";
     };
     // NOTE : Do not add default case in the switch above. In future if we get new cases for
     // Schema and miss them in the switch above we would like to get notified. Adding
diff --git a/pulsar-client-cpp/tests/CMakeLists.txt b/pulsar-client-cpp/tests/CMakeLists.txt
index 9e74de2597aeb..fe5f23582c70d 100644
--- a/pulsar-client-cpp/tests/CMakeLists.txt
+++ b/pulsar-client-cpp/tests/CMakeLists.txt
@@ -17,6 +17,21 @@
 # under the License.
 #
 
+if (NOT PROTOC_PATH)
+    set(PROTOC_PATH protoc)
+endif()
+
+set(LIB_AUTOGEN_DIR ${AUTOGEN_DIR}/tests)
+file(MAKE_DIRECTORY ${LIB_AUTOGEN_DIR})
+include_directories(${LIB_AUTOGEN_DIR})
+
+set(PROTO_DIR ${CMAKE_CURRENT_SOURCE_DIR}/../../pulsar-client/src/test/proto)
+set(PROTO_SOURCES ${LIB_AUTOGEN_DIR}/Test.pb.cc ${LIB_AUTOGEN_DIR}/ExternalTest.pb.cc)
+add_custom_command(
+    OUTPUT ${PROTO_SOURCES}
+    COMMAND ${PROTOC_PATH} -I ${PROTO_DIR} ${PROTO_DIR}/Test.proto ${PROTO_DIR}/ExternalTest.proto --cpp_out=${LIB_AUTOGEN_DIR})
+include_directories(${LIB_AUTOGEN_DIR})
+
 find_library(GMOCK_LIBRARY_PATH gmock)
 find_library(GTEST_LIBRARY_PATH gtest)
 find_library(GMOCKD_LIBRARY_PATH gmockd)
@@ -30,7 +45,7 @@ endif()
 
 file(GLOB TEST_SOURCES *.cc)
 
-add_executable(main ${TEST_SOURCES})
+add_executable(main ${TEST_SOURCES} ${PROTO_SOURCES})
 
 target_include_directories(main PRIVATE ${CMAKE_SOURCE_DIR}/lib ${AUTOGEN_DIR}/lib)
 
diff --git a/pulsar-client-cpp/tests/ProtobufNativeSchemaTest.cc b/pulsar-client-cpp/tests/ProtobufNativeSchemaTest.cc
new file mode 100644
index 0000000000000..3b75d4a47fea8
--- /dev/null
+++ b/pulsar-client-cpp/tests/ProtobufNativeSchemaTest.cc
@@ -0,0 +1,124 @@
+/**
+ * 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.
+ */
+#include <gtest/gtest.h>
+#include <pulsar/Client.h>
+#include <pulsar/ProtobufNativeSchema.h>
+#include <stdexcept>
+#include "Test.pb.h"  // generated from "pulsar-client/src/test/proto/Test.proto"
+
+using namespace pulsar;
+
+static std::string lookupUrl = "pulsar://localhost:6650";
+
+TEST(ProtobufNativeSchemaTest, testSchemaJson) {
+    const std::string expectedSchemaJson =
+        "{\"fileDescriptorSet\":"
+        "\"CtMDCgpUZXN0LnByb3RvEgVwcm90bxoSRXh0ZXJuYWxUZXN0LnByb3RvImUKClN1Yk1lc3NhZ2USCwoDZm9vGAEgASgJEgsKA2"
+        "JhchgCIAEoARo9Cg1OZXN0ZWRNZXNzYWdlEgsKA3VybBgBIAEoCRINCgV0aXRsZRgCIAEoCRIQCghzbmlwcGV0cxgDIAMoCSLlAQ"
+        "oLVGVzdE1lc3NhZ2USEwoLc3RyaW5nRmllbGQYASABKAkSEwoLZG91YmxlRmllbGQYAiABKAESEAoIaW50RmllbGQYBiABKAUSIQ"
+        "oIdGVzdEVudW0YBCABKA4yDy5wcm90by5UZXN0RW51bRImCgtuZXN0ZWRGaWVsZBgFIAEoCzIRLnByb3RvLlN1Yk1lc3NhZ2USFQ"
+        "oNcmVwZWF0ZWRGaWVsZBgKIAMoCRI4Cg9leHRlcm5hbE1lc3NhZ2UYCyABKAsyHy5wcm90by5leHRlcm5hbC5FeHRlcm5hbE1lc3"
+        "NhZ2UqJAoIVGVzdEVudW0SCgoGU0hBUkVEEAASDAoIRkFJTE9WRVIQAUItCiVvcmcuYXBhY2hlLnB1bHNhci5jbGllbnQuc2NoZW"
+        "1hLnByb3RvQgRUZXN0YgZwcm90bzMKoAEKEkV4dGVybmFsVGVzdC5wcm90bxIOcHJvdG8uZXh0ZXJuYWwiOwoPRXh0ZXJuYWxNZX"
+        "NzYWdlEhMKC3N0cmluZ0ZpZWxkGAEgASgJEhMKC2RvdWJsZUZpZWxkGAIgASgBQjUKJW9yZy5hcGFjaGUucHVsc2FyLmNsaWVudC"
+        "5zY2hlbWEucHJvdG9CDEV4dGVybmFsVGVzdGIGcHJvdG8z\",\"rootMessageTypeName\":\"proto.TestMessage\","
+        "\"rootFileDescriptorName\":\"Test.proto\"}";
+    const auto schemaInfo = createProtobufNativeSchema(::proto::TestMessage::GetDescriptor());
+
+    ASSERT_EQ(schemaInfo.getSchemaType(), pulsar::PROTOBUF_NATIVE);
+    ASSERT_TRUE(schemaInfo.getName().empty());
+    ASSERT_EQ(schemaInfo.getSchema(), expectedSchemaJson);
+    ASSERT_TRUE(schemaInfo.getProperties().empty());
+}
+
+TEST(ProtobufNativeSchemaTest, testAutoCreateSchema) {
+    const std::string topicPrefix = "ProtobufNativeSchemaTest-testAutoCreateSchema-";
+    Client client(lookupUrl);
+
+    const auto schemaInfo = createProtobufNativeSchema(::proto::TestMessage::GetDescriptor());
+    Producer producer;
+    ASSERT_EQ(ResultOk, client.createProducer(topicPrefix + "producer",
+                                              ProducerConfiguration().setSchema(schemaInfo), producer));
+    Consumer consumer;
+    ASSERT_EQ(ResultOk, client.subscribe(topicPrefix + "consumer", "my-sub",
+                                         ConsumerConfiguration().setSchema(schemaInfo), consumer));
+    client.close();
+}
+
+TEST(ProtobufNativeSchemaTest, testSchemaIncompatibility) {
+    const std::string topic = "ProtobufNativeSchemaTest-testSchemaIncompatibility";
+    Client client(lookupUrl);
+
+    Producer producer;
+    auto createProducerResult = [&](const google::protobuf::Descriptor* descriptor) {
+        return client.createProducer(
+            topic, ProducerConfiguration().setSchema(createProtobufNativeSchema(descriptor)), producer);
+    };
+
+    // Create the protobuf native schema automatically
+    ASSERT_EQ(ResultOk, createProducerResult(::proto::TestMessage::GetDescriptor()));
+    producer.close();
+
+    // Try to create producer with another protobuf generated class
+    ASSERT_EQ(ResultIncompatibleSchema,
+              createProducerResult(::proto::external::ExternalMessage::GetDescriptor()));
+
+    // Try to create producer with the original schema again
+    ASSERT_EQ(ResultOk, createProducerResult(::proto::TestMessage::GetDescriptor()));
+
+    // createProtobufNativeSchema() cannot accept a null descriptor
+    try {
+        createProducerResult(nullptr);
+    } catch (const std::invalid_argument& e) {
+        ASSERT_STREQ(e.what(), "descriptor is null");
+    }
+
+    client.close();
+}
+
+TEST(ProtobufNativeSchemaTest, testEndToEnd) {
+    const std::string topic = "ProtobufSchemaTest-testEndToEnd";
+    Client client(lookupUrl);
+
+    const auto schemaInfo = createProtobufNativeSchema(::proto::TestMessage::GetDescriptor());
+    Consumer consumer;
+    ASSERT_EQ(ResultOk,
+              client.subscribe(topic, "my-sub", ConsumerConfiguration().setSchema(schemaInfo), consumer));
+    Producer producer;
+    ASSERT_EQ(ResultOk,
+              client.createProducer(topic, ProducerConfiguration().setSchema(schemaInfo), producer));
+
+    // Send a message that is serialized from a ProtoBuf class
+    ::proto::TestMessage testMessage;
+    testMessage.set_testenum(::proto::TestEnum::FAILOVER);
+    std::string content(testMessage.ByteSizeLong(), '\0');
+    testMessage.SerializeToArray(const_cast<char*>(content.data()), content.size());
+    ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(content).build()));
+
+    // Receive a message and parse it to the ProtoBuf class
+    ::proto::TestMessage receivedTestMessage;
+    ASSERT_EQ(receivedTestMessage.testenum(), ::proto::TestEnum::SHARED);
+
+    Message msg;
+    ASSERT_EQ(ResultOk, consumer.receive(msg, 3000));
+    receivedTestMessage.ParseFromArray(msg.getData(), msg.getLength());
+    ASSERT_EQ(receivedTestMessage.testenum(), ::proto::TestEnum::FAILOVER);
+
+    client.close();
+}
diff --git a/pulsar-client-cpp/tests/SchemaTest.cc b/pulsar-client-cpp/tests/SchemaTest.cc
index e11069fb06b8a..1776460f895a2 100644
--- a/pulsar-client-cpp/tests/SchemaTest.cc
+++ b/pulsar-client-cpp/tests/SchemaTest.cc
@@ -36,6 +36,7 @@ TEST(SchemaTest, testSchema) {
     ProducerConfiguration producerConf;
     producerConf.setSchema(SchemaInfo(AVRO, "Avro", exampleSchema));
     res = client.createProducer("topic-avro", producerConf, producer);
+    ASSERT_EQ(res, ResultOk);
 
     // Check schema version
     ASSERT_FALSE(producer.getSchemaVersion().empty());
@@ -69,4 +70,6 @@ TEST(SchemaTest, testSchema) {
     consumerConf.setSchema(SchemaInfo(JSON, "Json", "{}"));
     res = client.subscribe("topic-avro", "sub-2", consumerConf, consumer);
     ASSERT_EQ(ResultIncompatibleSchema, res);
+
+    client.close();
 }
diff --git a/site2/docs/client-libraries-cpp.md b/site2/docs/client-libraries-cpp.md
index c495f19446766..847a80d2a6ef3 100644
--- a/site2/docs/client-libraries-cpp.md
+++ b/site2/docs/client-libraries-cpp.md
@@ -20,8 +20,8 @@ You need to install the following components before using the C++ client:
 
 * [CMake](https://cmake.org/)
 * [Boost](http://www.boost.org/)
-* [Protocol Buffers](https://developers.google.com/protocol-buffers/) 2.6
-* [libcurl](https://curl.haxx.se/libcurl/)
+* [Protocol Buffers](https://developers.google.com/protocol-buffers/) >= 3
+* [libcurl](https://curl.se/libcurl/)
 * [Google Test](https://github.com/google/googletest)
 
 ## Linux