[schemaregistry]ProtobufNative Schema Support

[hnail]

Fixes #7642
Fixes #7674

Motivation

Protobuf Native Schema Support : PIP-Add ProtobufNative Schema Support

This PR proposes to import a new protobuf-v3 schema based on protobuf native Descriptor, Current ProtobufSchema is based on AVRO schema, this Causes the following restrictions :

• Current ProtobufSchema based on avro-protobuf only works when JVM classloader can load Java implementation class of GeneratedMessageV3 . this is unfriendly for long-running server ( e.g. Presto ), restart server for update class is unfriendly.
• Describe protobuf schema by avro schema will cause losses of information, so based current ProtobufSchema can’t support AutoConsume by DynamicMessage.
• The amount of support avro language is less than support protobuf language.

In consideration of backward compatibility , we add a new schema type SchemaType.PROTOBUF_NATIVE base on protobuf-v3 native Descriptor instead of modify SchemaType.PROTOBUF, aim to support GenericProtobufNativeSchema and AutoConsumeSchema for PROTOBUF.

Modifications

Describe in PIP-Add ProtobufNative Schema Support

---

Usage Example

Create ProtobufNative schema Example

SchemaDefinition def =  SchemaDefinition.<PBMessage>builder()
                .withAlwaysAllowNull(true)
                .withPojo(PBMessage.class)
                .build();
Schema schema = Schema.PROTOBUFNATIVE(def);
admin.schemas().createSchema(topic, schema.getSchemaInfo());

Class-Based schema consumer Example

Consumer<PBMessage> consumer = client.newConsumer(Schema.PROTOBUFNATIVE(PBMessage.class)).topic(topic)
                .subscriptionName("my-subscription-name").subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
        while (!consumer.hasReachedEndOfTopic()){
            Message<PBMessage> msg = consumer.receive();
            PBMessage pBMessage =  msg.getValue();
            System.out.println("getPlatform : "+ pBMessage.getPlatform());
        }

ProtobufNative AUTO_CONSUME Example

 Consumer<GenericRecord> consumer = client.newConsumer(Schema.AUTO_CONSUME()).topic(topic)
                .subscriptionName("my-subscription-nameaaa").subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
        .subscribe();
        while (!consumer.hasReachedEndOfTopic()){
            Message<GenericRecord> msg = consumer.receive() ;
           // System.out.println("GenericRecord :"+msg.getValue().toString());
            GenericRecord genericRecord =  msg.getValue();
            System.out.println(genericRecord.getField("platform"));
            GenericProtobufNativeRecord genericProtobufNativeRecord = (GenericProtobufNativeRecord)genericRecord;
            DynamicMessage dynamicMessage = genericProtobufNativeRecord.getProtobufRecord();         
            String platform = dynamicMessage.getField(dynamicMessage.getDescriptorForType().findFieldByName("platform")).toString();
            System.out.println("platform : "+ platform);
        }

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: ( yes )
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (PIP-Add ProtobufNative Schema Support )

[sijie]

@hnail Can you send the PIP of adding native protobuf support to dev@ mailing list? So the community can review the PIP and we can add the PIP to Pulsar's wiki page.

[hnail]

@hnail Can you send the PIP of adding native protobuf support to dev@ mailing list? So the community can review the PIP and we can add the PIP to Pulsar's wiki page.

send <[DISCUSS] PIP-Add ProtobufNative Schema Support> to [email protected] done

[hnail]

/pulsarbot run-failure-checks

[hnail]

@sijie @codelipenghui Has there been progresses on this review ?

[sijie]

@hnail overall the change looks really good. I made a few comments. PTAL.

[codelipenghui]

Looks good to me, just left some minor comments.

[codelipenghui]

@congbobo184 Could you please help review this PR?

[hnail]

/pulsarbot run-failure-checks

[fjod]

I have a question about FileDescriptorSet field. In https://github.com/apache/pulsar/blob/c01b1eeda3221bdbf863bf0f3f8373e93d90adef/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java file there is an example test class. It is generated from Test.proto and ExternalMessage.proto files. The problem is that, no matter what I try with protoc , I can not get same FileDescriptorSet content. It's always slightly different. I tried csharp, java, cpp output, with or without --descriptor_set_out - I always get different byte array (and base64 string). And I can not create producer/consumer on apache pulsar client with FileDescriptorSet generated by me, I get com.google.protobuf.InvalidProtocolBufferException.invalidEndTag exception which as far as I googled tells that data is corrupted. When I try to create producer with FileDescriptorSet data from test, it works. What am I doing wrong while generating descriptors using protoc ?

[hnail]

I have a question about FileDescriptorSet field. In https://github.com/apache/pulsar/blob/c01b1eeda3221bdbf863bf0f3f8373e93d90adef/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaTest.java file there is an example test class. It is generated from Test.proto and ExternalMessage.proto files. The problem is that, no matter what I try with protoc , I can not get same FileDescriptorSet content. It's always slightly different. I tried csharp, java, cpp output, with or without --descriptor_set_out - I always get different byte array (and base64 string). And I can not create producer/consumer on apache pulsar client with FileDescriptorSet generated by me, I get com.google.protobuf.InvalidProtocolBufferException.invalidEndTag exception which as far as I googled tells that data is corrupted. When I try to create producer with FileDescriptorSet data from test, it works. What am I doing wrong while generating descriptors using protoc ?

hello , happy for your take notice of this pull request .
I test as your description with the two following test case :

• FileDescriptorSet build by JAVA Proto Class , same as ProtobufNativeSchemaUtils#serialize

// buid FileDescriptorSet byte[] by Class  
byte[] fileDescriptorSetBytesByClass = DescriptorProtos.FileDescriptorSet.newBuilder()
                .addFile(Service.ServiceRequest.getDescriptor().getFile().toProto())
                .build().toByteArray();  

DescriptorProtos.FileDescriptorSet fileDescriptorSetByClass = DescriptorProtos.FileDescriptorSet
                .parseFrom(fileDescriptorSetBytesByClass);

// print FileDescriptorSet string which buid by Class
System.out.println(new String(fileDescriptorSetByClass.toBuilder().build().toByteArray(),"utf-8"));

• FileDescriptorSet build by protoc command , same as your description:

// buid FileDescriptorSet byte[] by 'protoc --include_imports --descriptor_set_out Request.desc Request.proto
byte[] fileDescriptorSetBytesByProtoC = FileUtils.readFileToByteArray(new File("Request.desc"));

DescriptorProtos.FileDescriptorSet fileDescriptorSetByProtoC = DescriptorProtos.FileDescriptorSet
                .parseFrom(fileDescriptorSetBytesByProtoC);

// print FileDescriptorSet string which buid by ProtoC
System.out.println(new String(fileDescriptorSetByClass.toBuilder().build().toByteArray(),"utf-8"));

The two realize above works for me :

• The serialized FileDescriptorSet byte[] is slightly different as your description, I think maybe java proto class lack of information compare with proto file when compile java class .
• Deserialize FileDescriptorSet is by DescriptorProtos.FileDescriptorSet.parseFrom(bytes[]) in two way above ProtobufNativeSchemaUtils#deserialize . so , even though byte[] is slightly different , but at the same of working .
• The current realize is reference google-doc : protobuf v3 Self-describing Messages

---

so , I think the reason is new ObjectMapper().writeValueAsBytes(schemaData); which serialize byte[] to String , may be is 'UTF-8' or 'Base64' ?

[fjod]

I found the culprit here, I deleted important information from source .proto file like java_package and java_outer_classname, and protoc was generating different info which apache pulsar was not able to use.

[hnail]

I found the culprit here, I deleted important information from source .proto file like java_package and java_outer_classname, and protoc was generating different info which apache pulsar was not able to use.

I recommend you can use JAVA API to create Schema like this :

SchemaDefinition def =  SchemaDefinition.<ProtoMsg>builder()
                .withPojo(ProtoMsg.class)
                .build();
Schema schema = Schema.PROTOBUF_NATIVE(def);
admin.schemas().createSchema(topic, schema.getSchemaInfo());

The java_package and java_outer_classname restrictions which you mentioned may need check in other language client when create schema by POJO API , be appreciate if your have time to help pull a issue to describe the problem for future optimization .

[fjod]

The java_package and java_outer_classname restrictions which you mentioned may need check in other language client when create schema by POJO API , be appreciate if your have time to help pull a issue to describe the problem for future optimization .

I created issue here about csharp/java mismatch in generating Descriptors

protocolbuffers/protobuf#8425

[fjod]

@hnail I think some bug in deserialize method of ProtobufNativeSchemaUtils. I generated descriptor using probuf-net library, here it is CgpUZXN0LnByb3RvEgVwcm90byImCgpYWFhNZXNzYWdlEgsKA2ZvbxgBIAEoCRILCgNiYXIYAiABKAFCD6oCDG5hdGl2ZUNoZWNrMmIGcHJvdG8z . If you check it on https://protogen.marcgravell.com/decode , it decodes fine. Here is json file with some additional fields:

{"fileDescriptorSet":"ClMKClRlc3QucHJvdG8SC25hdGl2ZUNoZWNrIjAKClhYWE1lc3NhZ2USEAoDZm9vGAEgASgJUgNmb28SEAoDYmFyGAIgASgBUgNiYXJiBnByb3RvMw==","rootMessageTypeName":"proto.XXXMessage","rootFileDescriptorName":"Test.proto"}

When I try to send it to apache-pulsar server, I get null-pointer exception on line 104 at

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufNativeSchemaUtils.java

Line 104 in 2a1828c

//extract nested message

I created sample java app copy-pasting code from ProtobufNativeSchemaUtils. Here is simpler version of deserializer:

         File file;
         Descriptors.Descriptor descriptor;
         file = new File("D:\\serialized");
         byte[] fileContent = Files.readAllBytes(file.toPath());
         ProtobufNativeSchemaData schemaData = new ObjectMapper().readValue(fileContent, ProtobufNativeSchemaData.class);
         FileDescriptorSet fileDescriptorSet = FileDescriptorSet.parseFrom(schemaData.fileDescriptorSet);   
        Map<String, FileDescriptorProto> fileDescriptorProtoCache = new HashMap<>();
        Map<String, Descriptors.FileDescriptor> fileDescriptorCache = new HashMap<>();
        fileDescriptorSet.getFileList().forEach(fileDescriptorProto -> fileDescriptorProtoCache.put(fileDescriptorProto.getName(), fileDescriptorProto));        
        FileDescriptorProto rootFileDescriptorProto = fileDescriptorProtoCache.get("Test.proto");
        deserializeFileDescriptor(rootFileDescriptorProto, fileDescriptorCache, fileDescriptorProtoCache);
        Descriptors.FileDescriptor fileDescriptor = fileDescriptorCache.get("Test.proto");      
        System.out.print(fileDescriptor.toProto());

output is:

name: "Test.proto"
package: "nativeCheck"
message_type {
  name: "XXXMessage"
  field {
    name: "foo"
    number: 1
    label: LABEL_OPTIONAL
    type: TYPE_STRING
    json_name: "foo"
  }
  field {
    name: "bar"
    number: 2
    label: LABEL_OPTIONAL
    type: TYPE_DOUBLE
    json_name: "bar"
  }
}
syntax: "proto3"

So Descriptor is fine! But if I try to use code from the repo:

        Descriptors.Descriptor descriptor2;
        Descriptors.FileDescriptor fileDescriptor2 = fileDescriptorCache.get(schemaData.rootMessageTypeName);
        String package1 = fileDescriptor2.getPackage();
                //trim package
        String[] paths = StringUtils.removeFirst(schemaData.rootMessageTypeName, package1).replaceFirst("\\.", "").split("\\.");
        //extract root message
        descriptor2 = fileDescriptor2.findMessageTypeByName(paths[0]);
        //extract nested message
        for (int i = 1; i < paths.length; i++) {
            descriptor2 = descriptor2.findNestedTypeByName(paths[i]);
        }
        System.out.print(descriptor2.toProto());

I get nullpointer exception on line with getPackage(); - similar to apache-pulsar exception.

I guess our generated descriptor does not contain such "package" and fails therefore? It would be great if we can use descriptor from our client too ;)

[hnail]

@fjod Hi , Your rootMessageTypeName is incorrect , change to nativeCheck.XXXMessage , rootMessageTypeName must be full path , nativeCheck is your package path .

Simpler Test code :

String schemaDataBytes = "{\"fileDescriptorSet" +
                "\":\"ClMKClRlc3QucHJvdG8SC25hdGl2ZUNoZWNrIjAKClhYWE1lc3NhZ2USEAoDZm9vGAEgASgJUgNmb28SEAoDYmFyGAIgASgBUgNiYXJiBnByb3RvMw==\",\"rootMessageTypeName\":\"nativeCheck.XXXMessage\",\"rootFileDescriptorName\":\"Test.proto\"}";
        
Descriptors.Descriptor descriptor = ProtobufNativeSchemaUtils.deserialize(schemaDataBytes.getBytes());

System.out.println(descriptor);

It is works fine .