[C++] Add padding characters to base64 encoded protobuf native schema

pulsar-client-cpp/lib/ProtobufNativeSchema.cc

@@ -51,8 +51,15 @@ SchemaInfo createProtobufNativeSchema(const google::protobuf::Descriptor* descri
     std::vector<char> bytes(fileDescriptorSet.ByteSizeLong());
     fileDescriptorSet.SerializeToArray(bytes.data(), bytes.size());
 
+    std::string base64String{base64(bytes.data()), base64(bytes.data() + bytes.size())};
+    // Pulsar broker only supports decoding Base64 with padding so we need to add padding '=' here
+    const size_t numPadding = 4 - base64String.size() % 4;
+    for (size_t i = 0; i < numPadding; i++) {
+        base64String.push_back('=');
+    }
+
     const std::string schemaJson =
-        R"({"fileDescriptorSet":")" + std::string(base64(bytes.data()), base64(bytes.data() + bytes.size())) +
+        R"({"fileDescriptorSet":")" + base64String +
         R"(","rootMessageTypeName":")" + rootMessageTypeName +
         R"(","rootFileDescriptorName":")" + rootFileDescriptorName + R"("})";
 

pulsar-client-cpp/tests/CMakeLists.txt

@@ -30,6 +30,15 @@ set(PROTO_SOURCES ${LIB_AUTOGEN_DIR}/Test.pb.cc ${LIB_AUTOGEN_DIR}/ExternalTest.
 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})
+
+set(PROTO_SOURCE_PADDING ${LIB_AUTOGEN_DIR}/PaddingDemo.pb.cc)
+add_custom_command(
+    OUTPUT ${PROTO_SOURCE_PADDING}
+    COMMAND ${PROTOC_PATH} -I . ./PaddingDemo.proto --cpp_out=${LIB_AUTOGEN_DIR}
+    WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
+
+set(PROTO_SOURCES ${PROTO_SOURCES} ${PROTO_SOURCE_PADDING})
+
 include_directories(${LIB_AUTOGEN_DIR})
 
 find_library(GMOCK_LIBRARY_PATH gmock)

pulsar-client-cpp/tests/PaddingDemo.proto

@@ -0,0 +1,26 @@
+/**
+ * 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 padding.demo;
+
+message Person {
+  string name = 1;
+  optional int32 id = 2;
+}

pulsar-client-cpp/tests/ProtobufNativeSchemaTest.cc

@@ -20,6 +20,7 @@
 #include <pulsar/Client.h>
 #include <pulsar/ProtobufNativeSchema.h>
 #include <stdexcept>
+#include "PaddingDemo.pb.h"
 #include "Test.pb.h"  // generated from "pulsar-client/src/test/proto/Test.proto"
 
 using namespace pulsar;
@@ -122,3 +123,21 @@ TEST(ProtobufNativeSchemaTest, testEndToEnd) {
 
     client.close();
 }
+
+TEST(ProtobufNativeSchemaTest, testBase64WithPadding) {
+    const auto schemaInfo = createProtobufNativeSchema(::padding::demo::Person::GetDescriptor());
+    const auto schemaJson = schemaInfo.getSchema();
+    size_t pos = schemaJson.find(R"(","rootMessageTypeName":)");
+    ASSERT_NE(pos, std::string::npos);
+    ASSERT_TRUE(pos > 0);
+    ASSERT_EQ(schemaJson[pos - 1], '=');  // the tail of fileDescriptorSet is a padding character
+
+    Client client(lookupUrl);
+
+    const std::string topic = "ProtobufSchemaTest-testBase64WithPadding";
+    Producer producer;
+    ASSERT_EQ(ResultOk,
+              client.createProducer(topic, ProducerConfiguration().setSchema(schemaInfo), producer));
+
+    client.close();
+}