Create c++ producer/consumer with protobuf schema

[adg25]

Expected behavior

I would like to use the schema registry for my c++ consumers/producers with protobuf schemas.
There are examples/test for AVRO and Json but not for protobuf and I'm seeing failures creating the producer/consumer.

Actual behavior

c++ test app output:

2021-03-16 15:31:46.224 WARN  [140638990731008] ClientConnection:959 | [127.0.0.1:35056 -> 127.0.0.1:6650] Received error response from server: IncompatibleSchema -- req_id: 0
2021-03-16 15:31:46.224 ERROR [140638990731008] ConsumerImpl:257 | [persistent://public/default/schema-publisher23, adg_sub, 0] Failed to create consumer: IncompatibleSchema

pulsar-standalone container output:

pulsar-standalone    | 15:41:43.175 [ForkJoinPool.commonPool-worker-1] WARN  org.apache.pulsar.broker.service.ServerCnx - [/10.0.2.2:45928][persistent://public/default/schema-publisher23][adg_sub] Failed to create consumer: Invalid schema definition data for PROTOBUF schema

Steps to reproduce

Code snippet:

void createConsumer()
{
    /*
        message MyProto{
            int a = 1;
            double b = 2;
        }
    */
  ::pulsar::Client client("pulsar://0.0.0.0:6650");
  ::pulsar::ConsumerConfiguration config;
  MyProto msg{};
  msg.set_a(22);
  msg.set_b(1.0);
  std::string msgJson;
  google::protobuf::util::MessageToJsonString(msg, &msgJson);
  ::pulsar::SchemaInfo schema(::pulsar::SchemaType::PROTOBUF, "Protobuf", msgJson);
  config.setSchema(schema);
  ::pulsar::Consumer consumer;
  auto result = client.subscribe("persistent://public/default/schema-publisher23", "adg_sub",
                                 config, consumer);
}

System configuration

Using pulsar-standalone docker image

[codelipenghui]

@BewareMyPower Could you please help take a look at this issue?

[BewareMyPower]

Currently AVRO, JSON and PROTOBUF (not PROTOBUF_NATIVE) share the same schema validator. See

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidator.java

Lines 37 to 43 in 7d3317e

	static void validateSchemaData(SchemaData schemaData) throws InvalidSchemaDataException {
	switch (schemaData.getType()) {
	case AVRO:
	case JSON:
	case PROTOBUF:
	StructSchemaDataValidator.of().validate(schemaData);
	break;

Therefore, following code works well.

#include <iostream>
#include <pulsar/Client.h>
using namespace pulsar;

int main() {
    Client client("pulsar://localhost:6650");

    std::string exampleSchema =
        R"({"type":"record","name":"Example","namespace":"test","fields":[{"name":"a","type":"int"},{"name":"b","type":"double"}]})";

    ProducerConfiguration producerConfig;
    SchemaInfo schema(SchemaType::PROTOBUF, "Protobuf", exampleSchema);
    producerConfig.setSchema(schema);
    Producer producer;
    Result result = client.createProducer("my-topic", producerConfig, producer);
    if (result != ResultOk) {
        std::cout << "Failed to create producer: " << result << std::endl;
        return -1;
    }

    client.close();
    return 0;
}

The logs is

2021-03-18 14:30:45.723 INFO  [0x1141d0e00] ConnectionPool:85 | Created connection for pulsar://localhost:6650
2021-03-18 14:30:45.724 INFO  [0x700009a0a000] ClientConnection:372 | [[::1]:50989 -> [::1]:6650] Connected to broker
2021-03-18 14:30:45.749 INFO  [0x700009a0a000] HandlerBase:54 | [persistent://public/default/my-topic, ] Getting connection from pool
2021-03-18 14:30:46.260 INFO  [0x700009a0a000] ProducerImpl:176 | [persistent://public/default/my-topic, ] Created producer on broker [[::1]:50989 -> [::1]:6650] 
2021-03-18 14:30:46.260 INFO  [0x1141d0e00] ClientImpl:483 | Closing Pulsar client
2021-03-18 14:30:46.260 INFO  [0x1141d0e00] ProducerImpl:581 | [persistent://public/default/my-topic, standalone-0-0] Closing producer for topic persistent://public/default/my-topic
2021-03-18 14:30:46.264 INFO  [0x700009a0a000] ProducerImpl:624 | [persistent://public/default/my-topic, standalone-0-0] Closed producer
2021-03-18 14:30:46.264 INFO  [0x700009a0a000] ClientConnection:1446 | [[::1]:50989 -> [::1]:6650] Connection closed
2021-03-18 14:30:46.264 INFO  [0x1141d0e00] ProducerImpl:560 | Producer - [persistent://public/default/my-topic, standalone-0-0] , [batchMessageContainer = { BatchMessageContainer [size = 0] [bytes = 0] [maxSize = 1000] [maxBytes = 131072] [topicName = persistent://public/default/my-topic] [numberOfBatchesSent_ = 0] [averageBatchSize_ = 0] }]
2021-03-18 14:30:46.265 INFO  [0x1141d0e00] ClientConnection:261 | [[::1]:50989 -> [::1]:6650] Destroyed connection

So the problem is with your JSON string from google::protobuf::util::MessageToJsonString. Could you print the msgJson?

[adg25]

Yes that worked for me as well! MessageToJsonString will not work for what I need here

[codelipenghui]

The issue had no activity for 30 days, mark with Stale label.

[tisonkun]

Closed as stale.

The development of the C++ client has been permanently moved to https://github.com/apache/pulsar-client-cpp. If it's still relevant, please open an issue there.