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

…#11492)

### Motivation

#11388 added support for protobuf native schema in C++ client. The implementation serializes a protobuf generated class' descriptor to a Base64 encoded string. However, it doesn't add the padding characters while the broker side requires the padding characters. Therefore, if the Base64 encoded string needs the padding characters, the broker will fail to deserialize:

> com.fasterxml.jackson.databind.exc.InvalidFormatException: Cannot deserialize value of type `byte[]` from String "": Unexpected end of base64-encoded String: base64 variant 'MIME-NO-LINEFEEDS' expects padding (one or more '=' characters) at the end. This Base64Variant might have been incorrectly configured

See https://en.wikipedia.org/wiki/Base64#Decoding_Base64_with_padding for the Base64 with padding.

### Modifications

- Add the padding `=` characters if the Base64 encoded string's length is not a multiple of four.
- Add a `PaddingDemo.proto` to test the above case to ensure that broker can validate the schema string.

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();
+}