BE: Fix protobuf map support (#3683)

* ISSUE-3674: Setting untyped object schema for "map" protobuf fields

* com.provectus.kafka.ui.util.jsonschema classes visibility refactoring

* kafka-ui-serdes.yaml changes rolled back

---------

Co-authored-by: iliax <ikuramshin@provectus.com>

documentation/compose/proto/values.proto

@@ -9,4 +9,6 @@ message MySpecificTopicValue {
 message MyValue {
   int32 version = 1;
   string payload = 2;
+  map<int32, string> intToStringMap = 3;
+  map<string, MyValue> strToObjMap  = 4;
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/AnyFieldSchema.java

@@ -4,9 +4,9 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 // Specifies field that can contain any kind of value - primitive, complex and nulls
-public class AnyFieldSchema implements FieldSchema {
+class AnyFieldSchema implements FieldSchema {
 
-  public static AnyFieldSchema get() {
+  static AnyFieldSchema get() {
     return new AnyFieldSchema();
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/ArrayFieldSchema.java

@@ -4,10 +4,10 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
-public class ArrayFieldSchema implements FieldSchema {
+class ArrayFieldSchema implements FieldSchema {
   private final FieldSchema itemsSchema;
 
-  public ArrayFieldSchema(FieldSchema itemsSchema) {
+  ArrayFieldSchema(FieldSchema itemsSchema) {
     this.itemsSchema = itemsSchema;
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/EnumJsonType.java

@@ -7,10 +7,10 @@ import java.util.List;
 import java.util.Map;
 
 
-public class EnumJsonType extends JsonType {
+class EnumJsonType extends JsonType {
   private final List<String> values;
 
-  public EnumJsonType(List<String> values) {
+  EnumJsonType(List<String> values) {
     super(Type.ENUM);
     this.values = values;
   }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/FieldSchema.java

@@ -3,6 +3,6 @@ package com.provectus.kafka.ui.util.jsonschema;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
-public interface FieldSchema {
+interface FieldSchema {
   JsonNode toJsonNode(ObjectMapper mapper);
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/JsonType.java

@@ -4,7 +4,7 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.Map;
 
-public abstract class JsonType {
+abstract class JsonType {
 
   protected final Type type;
 
@@ -12,13 +12,13 @@ public abstract class JsonType {
     this.type = type;
   }
 
-  public Type getType() {
+  Type getType() {
     return type;
   }
 
-  public abstract Map<String, JsonNode> toJsonNode(ObjectMapper mapper);
+  abstract Map<String, JsonNode> toJsonNode(ObjectMapper mapper);
 
-  public enum Type {
+  enum Type {
     NULL,
     BOOLEAN,
     OBJECT,

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/MapFieldSchema.java

@@ -2,21 +2,27 @@ package com.provectus.kafka.ui.util.jsonschema;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.BooleanNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.node.TextNode;
+import javax.annotation.Nullable;
 
-public class MapFieldSchema implements FieldSchema {
-  private final FieldSchema itemSchema;
+class MapFieldSchema implements FieldSchema {
+  private final @Nullable FieldSchema itemSchema;
 
-  public MapFieldSchema(FieldSchema itemSchema) {
+  MapFieldSchema(@Nullable FieldSchema itemSchema) {
     this.itemSchema = itemSchema;
   }
 
+  MapFieldSchema() {
+    this(null);
+  }
+
   @Override
   public JsonNode toJsonNode(ObjectMapper mapper) {
     final ObjectNode objectNode = mapper.createObjectNode();
     objectNode.set("type", new TextNode(JsonType.Type.OBJECT.getName()));
-    objectNode.set("additionalProperties", itemSchema.toJsonNode(mapper));
+    objectNode.set("additionalProperties", itemSchema != null ? itemSchema.toJsonNode(mapper) : BooleanNode.TRUE);
     return objectNode;
   }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/ObjectFieldSchema.java

@@ -9,24 +9,24 @@ import java.util.stream.Collectors;
 import reactor.util.function.Tuple2;
 import reactor.util.function.Tuples;
 
-public class ObjectFieldSchema implements FieldSchema {
+class ObjectFieldSchema implements FieldSchema {
 
-  public static final ObjectFieldSchema EMPTY = new ObjectFieldSchema(Map.of(), List.of());
+  static final ObjectFieldSchema EMPTY = new ObjectFieldSchema(Map.of(), List.of());
 
   private final Map<String, FieldSchema> properties;
   private final List<String> required;
 
-  public ObjectFieldSchema(Map<String, FieldSchema> properties,
+  ObjectFieldSchema(Map<String, FieldSchema> properties,
                            List<String> required) {
     this.properties = properties;
     this.required = required;
   }
 
-  public Map<String, FieldSchema> getProperties() {
+  Map<String, FieldSchema> getProperties() {
     return properties;
   }
 
-  public List<String> getRequired() {
+  List<String> getRequired() {
     return required;
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/OneOfFieldSchema.java

@@ -5,11 +5,10 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.List;
 import java.util.stream.Collectors;
 
-public class OneOfFieldSchema implements FieldSchema {
+class OneOfFieldSchema implements FieldSchema {
   private final List<FieldSchema> schemaList;
 
-  public OneOfFieldSchema(
-      List<FieldSchema> schemaList) {
+  OneOfFieldSchema(List<FieldSchema> schemaList) {
     this.schemaList = schemaList;
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/ProtobufSchemaConverter.java

@@ -94,6 +94,9 @@ public class ProtobufSchemaConverter implements JsonSchemaConverter<Descriptors.
     if (wellKnownTypeSchema.isPresent()) {
       return wellKnownTypeSchema.get();
     }
+    if (field.isMapField()) {
+      return new MapFieldSchema();
+    }
     final JsonType jsonType = convertType(field);
     FieldSchema fieldSchema;
     if (jsonType.getType().equals(JsonType.Type.OBJECT)) {
@@ -149,67 +152,47 @@ public class ProtobufSchemaConverter implements JsonSchemaConverter<Descriptors.
   }
 
   private JsonType convertType(Descriptors.FieldDescriptor field) {
-    switch (field.getType()) {
-      case INT32:
-      case FIXED32:
-      case SFIXED32:
-      case SINT32:
-        return new SimpleJsonType(
-            JsonType.Type.INTEGER,
-            Map.of(
-                "maximum", IntNode.valueOf(Integer.MAX_VALUE),
-                "minimum", IntNode.valueOf(Integer.MIN_VALUE)
-            )
-        );
-      case UINT32:
-        return new SimpleJsonType(
-            JsonType.Type.INTEGER,
-            Map.of(
-                "maximum", LongNode.valueOf(UnsignedInteger.MAX_VALUE.longValue()),
-                "minimum", IntNode.valueOf(0)
-            )
-        );
+    return switch (field.getType()) {
+      case INT32, FIXED32, SFIXED32, SINT32 -> new SimpleJsonType(
+          JsonType.Type.INTEGER,
+          Map.of(
+              "maximum", IntNode.valueOf(Integer.MAX_VALUE),
+              "minimum", IntNode.valueOf(Integer.MIN_VALUE)
+          )
+      );
+      case UINT32 -> new SimpleJsonType(
+          JsonType.Type.INTEGER,
+          Map.of(
+              "maximum", LongNode.valueOf(UnsignedInteger.MAX_VALUE.longValue()),
+              "minimum", IntNode.valueOf(0)
+          )
+      );
       //TODO: actually all *64 types will be printed with quotes (as strings),
       // see JsonFormat::printSingleFieldValue for impl. This can cause problems when you copy-paste from messages
       // table to `Produce` area - need to think if it is critical or not.
-      case INT64:
-      case FIXED64:
-      case SFIXED64:
-      case SINT64:
-        return new SimpleJsonType(
-            JsonType.Type.INTEGER,
-            Map.of(
-                "maximum", LongNode.valueOf(Long.MAX_VALUE),
-                "minimum", LongNode.valueOf(Long.MIN_VALUE)
-            )
-        );
-      case UINT64:
-        return new SimpleJsonType(
-            JsonType.Type.INTEGER,
-            Map.of(
-                "maximum", new BigIntegerNode(UnsignedLong.MAX_VALUE.bigIntegerValue()),
-                "minimum", LongNode.valueOf(0)
-            )
-        );
-      case MESSAGE:
-      case GROUP:
-        return new SimpleJsonType(JsonType.Type.OBJECT);
-      case ENUM:
-        return new EnumJsonType(
-            field.getEnumType().getValues().stream()
-                .map(Descriptors.EnumValueDescriptor::getName)
-                .collect(Collectors.toList())
-        );
-      case BYTES:
-      case STRING:
-        return new SimpleJsonType(JsonType.Type.STRING);
-      case FLOAT:
-      case DOUBLE:
-        return new SimpleJsonType(JsonType.Type.NUMBER);
-      case BOOL:
-        return new SimpleJsonType(JsonType.Type.BOOLEAN);
-      default:
-        return new SimpleJsonType(JsonType.Type.STRING);
-    }
+      case INT64, FIXED64, SFIXED64, SINT64 -> new SimpleJsonType(
+          JsonType.Type.INTEGER,
+          Map.of(
+              "maximum", LongNode.valueOf(Long.MAX_VALUE),
+              "minimum", LongNode.valueOf(Long.MIN_VALUE)
+          )
+      );
+      case UINT64 -> new SimpleJsonType(
+          JsonType.Type.INTEGER,
+          Map.of(
+              "maximum", new BigIntegerNode(UnsignedLong.MAX_VALUE.bigIntegerValue()),
+              "minimum", LongNode.valueOf(0)
+          )
+      );
+      case MESSAGE, GROUP -> new SimpleJsonType(JsonType.Type.OBJECT);
+      case ENUM -> new EnumJsonType(
+          field.getEnumType().getValues().stream()
+              .map(Descriptors.EnumValueDescriptor::getName)
+              .collect(Collectors.toList())
+      );
+      case BYTES, STRING -> new SimpleJsonType(JsonType.Type.STRING);
+      case FLOAT, DOUBLE -> new SimpleJsonType(JsonType.Type.NUMBER);
+      case BOOL -> new SimpleJsonType(JsonType.Type.BOOLEAN);
+    };
   }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/RefFieldSchema.java

@@ -4,10 +4,10 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.TextNode;
 
-public class RefFieldSchema implements FieldSchema {
+class RefFieldSchema implements FieldSchema {
   private final String ref;
 
-  public RefFieldSchema(String ref) {
+  RefFieldSchema(String ref) {
     this.ref = ref;
   }
 
@@ -16,7 +16,7 @@ public class RefFieldSchema implements FieldSchema {
     return mapper.createObjectNode().set("$ref", new TextNode(ref));
   }
 
-  public String getRef() {
+  String getRef() {
     return ref;
   }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/SimpleFieldSchema.java

@@ -3,10 +3,10 @@ package com.provectus.kafka.ui.util.jsonschema;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
-public class SimpleFieldSchema implements FieldSchema {
+class SimpleFieldSchema implements FieldSchema {
   private final JsonType type;
 
-  public SimpleFieldSchema(JsonType type) {
+  SimpleFieldSchema(JsonType type) {
     this.type = type;
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/SimpleJsonType.java

@@ -6,15 +6,15 @@ import com.fasterxml.jackson.databind.node.TextNode;
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 
-public class SimpleJsonType extends JsonType {
+class SimpleJsonType extends JsonType {
 
   private final Map<String, JsonNode> additionalTypeProperties;
 
-  public SimpleJsonType(Type type) {
+  SimpleJsonType(Type type) {
     this(type, Map.of());
   }
 
-  public SimpleJsonType(Type type, Map<String, JsonNode> additionalTypeProperties) {
+  SimpleJsonType(Type type, Map<String, JsonNode> additionalTypeProperties) {
     super(type);
     this.additionalTypeProperties = additionalTypeProperties;
   }

kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/jsonschema/ProtobufSchemaConverterTest.java

@@ -59,8 +59,10 @@ class ProtobufSchemaConverterTest {
                 TestMsg outer_ref = 2;
                 EmbeddedMsg self_ref = 3;
             }
-        }""";
 
+            map<int32, string> intToStringMap = 21;
+            map<string, EmbeddedMsg> strToObjMap  = 22;
+        }""";
 
     String expectedJsonSchema = """
         {
@@ -109,7 +111,9 @@ class ProtobufSchemaConverterTest {
                         "v2": { "type": [ "number", "string", "object", "array", "boolean", "null" ] },
                         "uint32_w_field": { "type": "integer", "maximum": 4294967295, "minimum": 0 },
                         "bool_w_field": { "type": "boolean" },
-                        "uint64_w_field": { "type": "integer", "maximum": 18446744073709551615, "minimum": 0 }
+                        "uint64_w_field": { "type": "integer", "maximum": 18446744073709551615, "minimum": 0 },
+                        "strToObjMap": { "type": "object", "additionalProperties": true },
+                        "intToStringMap": { "type": "object", "additionalProperties": true }
                     }
                 },
                 "test.TestMsg.EmbeddedMsg": {