ISSUE-849: Use a map between topics and message-names when using ProtobufFile

[zarezadeh]

• Breaking change?

What changes did you make? Added the ability to specify separate message types for each topic when using ProtobufFile for deserialization. The configuration protobufMessageName is now a map between the topic name and the message name. The key _default is used for specifying the message name for all other topics not defined explicitly.

Is there anything you'd like reviewers to focus on?

How Has This Been Tested? (put an "X" next to an item)

• No need to
• Manually(please, describe, when necessary)
• Unit checks
• Integration checks
• Covered by existing automation

Checklist (put an "X" next to an item, otherwise PR will fail)

• I have performed a self-review of my own code
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation (e.g. ENVIRONMENT VARIABLES)
• My changes generate no new warnings(e.g. Sonar is happy)
• I have added tests that prove my fix is effective or that my feature works
• New and existing unit tests pass locally with my changes
• Any dependent changes have been merged

Check out Contributing and Code of Conduct

[iliax]

@zarezadeh thank you for PR!
Here some improvements to consider:

1. Lets keep configs backward-compatible: use protobufMessageName property value as default message and protobufTopicMessageName(or protobufMessageNameByTopic) map for explicit topic mapping
2. please add mapping check in ProtobufFileRecordSerDe::serialize method:

   kafka-ui/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/ProtobufFileRecordSerDe.java

   Line 81 in c19b9a6

   DynamicMessage.Builder builder = protobufSchema.newMessageBuilder();

3. In tests please add manual messages creations rather than Base64. Here is a sample how you can do it:

DynamicMessage.Builder builder = protobufSchema.newMessageBuilder("messageNameHere");
JsonFormat.parser().merge("{ \"name\": \"My Name\", \"email\": \"test@test.com\" }", builder);
byte[] bytes =  builder.build().toByteArray();

[iliax]

see #854 (comment)

[zarezadeh]

@iliax thanks for your review, I applied your suggestions in the code.
But I think there's still one case that we should take into account: What if the provided messageName for a given topic does not exist in the protobuf definition? Should we handle it or is it the user's responsibility to provide proper configuration?

[iliax]

@iliax thanks for your review, I applied your suggestions in the code.
But I think there's still one case that we should take into account: What if the provided messageName for a given topic does not exist in the protobuf definition? Should we handle it or is it the user's responsibility to provide proper configuration?

@zarezadeh Yes, it is a good point. I think we definitely should not fall with NPE when provided name not found. We should either check names in constructor ( toDescriptor(..) != null) or in serialize/derialize. (requireNonNull(..))

[iliax]

see comment above

[germanosin]

LGTM

[iliax]

@zarezadeh thank you for PR!