Support ANY field in Protobuf descriptors

[xakassi]

This a fix for a bug described in PR 529.

Any field is now supported for Protobuf descriptors. Multiple layer complex objects with Any fields on any layers are also supported using recursion. Please, see unit tests for examples.

The principle of operation is as follows. If a message contains a field with type Any, we can find out an actual field type at the runtime from the message, get descriptors for this field type and add them to TypeRegistry. To get the descriptors for this field type I pre-built a map with pairs - for all types provided in all descriptors.

Hi, @tchiotludo , @jorgheymans ! Please, take a look at this PR.
@jorgheymans I hope your case is fully covered now, please, test on your data.

[jorgheymans]

Going to test this out hopefully today, feedback soon.

[jorgheymans]

Looking at this, the proto serde is not working and i'm not seeing any error messages. This leads me to believe it's a configuration error. Can you confirm this config should be correct:

    deserialization:
      protobuf:
        descriptors-folder: "/opt/akhq/descriptors"
        topics-mapping:
          - topic-regex: "cc.*"
            descriptor-file: "cc-message-proto.protobin"
            value-message-type: "cc.message.CCMessage"
          - topic-regex: "han.*"
            descriptor-file: "han-cdc-proto.protobin"
            value-message-type: "han.message.HanMessage"

Note that the descriptor files are not self-containing as it doesn't make sense once you have Any. So the /opt/akhq/descriptors directory contains many protobin files, I am assuming you scan them all ?

So basically we have KafkaMessage.proto

message KafkaMessage {
    ..........
    Any payload
}

And then many payload types, each type has its own protobin file. All these protobin files are in the descriptors-folder.

[xakassi]

Could you, please, provide the whole config file? Maybe you have placed deserialization block in the wrong place? It should be under your Kafka cluster on the same level as its "properties". I will show you my whole config once I'm at the computer.

[xakassi]

Hi, @jorgheymans !
This is my whole config:

akhq:
  connections:
    kafka:
      properties:
        bootstrap.servers: "localhost:9094"
      deserialization:
        protobuf:
          descriptors-folder: "D:\\protobuf_desc"
          topics-mapping:
            - topic-regex: "test.*"
              descriptor-file: "streaming.desc"
              key-message-type: "Row"
              value-message-type: "Envelope"

It should not be a problem that descriptor files are not self-containing. I believe it's just configuration issue since there are no any error messages. Seems like deserialization is put in the wrong place and is not recognized at all.

[jorgheymans]

Indeed the config block was placed under topics, i have corrected it and as soon as I'm able to test this again i'll let you know, may take a few days with all the holidays going on.

[tchiotludo]

@jorgheymans : please tell me what you test is going on.
I would like to release a new AKHQ version will a full support of protobuf (v1).
Thanks for your time @xakassi & @jorgheymans !

[jorgheymans]

It won't be before next week, full holiday mode here sorry !

…

On Sun, Dec 27, 2020 at 10:15 PM Ludovic DEHON ***@***.***> wrote: @jorgheymans <https://github.com/jorgheymans> : please tell me what you test is going on. I would like to release a new AKHQ version will a full support of protobuf (v1). Thanks for your time @xakassi <https://github.com/xakassi> & @jorgheymans <https://github.com/jorgheymans> ! — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#543 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AABPKAA6OA3LE54ZWC55EJDSW6PXFANCNFSM4U6IWCVA> .

[tchiotludo]

@jorgheymans : have a good holidays !
see you next week !

[xakassi]

Hi, @jorgheymans !
Did you have any chance to test it yet?

[mdkouki]

Hi,following Jorg inputs test done and we got an exception here I share with you what we got :

cordially.

[xakassi]

Hi!
Are your descriptor files ok? It's an exception from standard function FileDescriptorSet.parseFrom(descriptorFile), it cannot parse your descriptor file.
Could you provide more info about your descriptor files and data? Maybe some fake data to reproduce.

[jorgheymans]

It's missing the google.protobuf.timestamp descriptor, this is part of the well known types https://developers.google.com/protocol-buffers/docs/reference/csharp/class/google/protobuf/well-known-types/timestamp I guess we should generate a separate descriptor for these internal types as well ?

…

On Mon, Jan 11, 2021 at 1:12 PM Taisiia Goltseva ***@***.***> wrote: Hi! Are your descriptor files ok? It's an exception from standard function FileDescriptorSet.parseFrom(descriptorFile), it cannot parse your descriptor file. Could you provide more info about your descriptor files and data? Maybe some fake data to reproduce. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#543 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AABPKAHBJ4SCLXEMTDBZFSTSZLTLPANCNFSM4U6IWCVA> .

[xakassi]

I have added a unit test for an object with Timestamp field. @jorgheymans , please, see the unit test deserializeFilm(). Film object now looks like this:

syntax = "proto3";
package org.akhq.utils;

import "google/protobuf/timestamp.proto";

option java_package = "org.akhq.utils";
option java_outer_classname = "FilmProto";

message Film {
  string name = 1;
  string producer = 2;
  int32 release_year = 3;
  int32 duration = 4;
  repeated string starring = 5;
  google.protobuf.Timestamp timestamp = 6;
}

And it deserialized successfully.
Do your objects looks like this also? Or I have specified Timestamp in a different way?

[jorgheymans]

The error surfaces if you have a timestamp field in a type that is defined as Any.

[xakassi]

Yes, indeed. But in my case (when I test) I get other error in a different function (not here FileDescriptorSet.parseFrom(descriptorFile)).

But anyway my current approach of collecting necessary descriptors is incomplete. And it would be better to just add all custom descriptors from a specified folder and add all well-known-descriptors.

The one question is - how to get all well-known-descriptors. @jorgheymans could you, please, tell me how you implemented that? How do you get all well-known-descriptors? It would be easier and faster for me :)

[xakassi]

@jorgheymans
Maybe you just call something like that for every well-known-type?

com.google.protobuf.StringValue.getDescriptor();

[geertvb]

@xakassi The .proto files for the well-known-descriptors can be found in the com.google.protobuf:protobuf-java jar under the folder google/protobuf. You can compile these proto definitions and generate a single protobin file. We have a maven pom file to do that.
Once you have this protobin file, it can be loaded and used like any other file descriptor set.

[xakassi]

Thanks, @geertvb ! I will try this approach!

[xakassi]

I had some troubles with making single DescriptorSet, and I have implemented it in a different way. I think, it's better in my case.

Please, guys, try your use case again with my new fixed code! @jorgheymans

[geertvb]

@xakassi Initially we also hard coded the well defined types like that but it is less flexible. Besides that we needed the protobin files for another project where we can't use that trick because we aren't using the google libraries.

[xakassi]

@geertvb You can put any additional protobin files in descriptors-folder, Descriptors will be extracted from them and will be used for fields with Any type if needed.

[dert1714]

Hi, @tchiotludo!
Do you have any another assumtion to merge this PR except unresolved conflicts?

[tchiotludo]

at @dert1714, I don't have a lot of knowledge of protobuf.
I resolved the conflict, If you can test if it's working I could merge it

[tchiotludo]

@dert1714 build failed due to incorrect unit that need to be fixed before merging

[stoibazarov]

I was able to fix this error by building proto.desc file with argument protoc --include_imports