Repository: provectus/kafka-ui
Issue: Glue serde avro to json deserialization includes namespaces and union types 
Labels: type/bug
scope/backend
status/accepted
area/serde

Originally reported in #3224 , split into a separate issue following the discussion in #3235 

When the glue serde deserializes to json from avro, it includes the record namespaces and types in the case of union. This is the first time I'm encountering the behaviour since the python deserializer or the one used in kafka-connect don't follow this behavior

### Example:
Original msg:
```json
{"name": {"first": "ron", "last": "serruya", "full": "ron serruya"}, "ids1": [5,6], "ids2": ["abc", 123]}
``` 
schema used:
```
{
  "type": "record",
  "name": "generation",
  "namespace": "top_level",
  "fields": [
    {
      "name": "name",
      "type": [
        {
          "type": "record",
          "name": "name",
          "namespace": "top_level.generation",
          "fields": [
            {
              "name": "raw",
              "type": [
                "string",
                "null"
              ]
            },
            {
              "name": "first",
              "type": "string"
            },
            {
              "name": "full",
              "type": "string"
            },
            {
              "name": "last",
              "type": ["string"]
            }
          ]
        },
        "null"
      ]
    },
    {
      "name": "ids1",
      "type": {"type": "array", "items": "int"}
    },
    {
      "name": "ids2",
      "type": {"type": "array", "items": ["string", "int"]}
    }
  ]
}
```

base64 encoded avro msg (just the msg, without the glue-related bytes at the start)
`AAIGcm9uFnJvbiBzZXJydXlhAA5zZXJydXlhBAoMAAQABmFiYwL2AQA=`

The current glue deserializer shows this msg as:
```
{
  "name": {
    "top_level.generation.name": {
      "raw": null,
      "first": "ron",
      "full": "ron serruya",
      "last": {
        "string": "serruya"
      }
    }
  },
  "ids1": [
    5,
    6
  ],
  "ids2": [
    {
      "string": "abc"
    },
    {
      "int": 123
    }
  ]
}
```
As you can see it adds `string`, `int`, or the record namespace `top_level.generation.name`

I fixed this issue locally by adding this line: `encoder.setIncludeNamespace(false);` in the [avroRecordToJson method](https://github.com/provectus/kafkaui-glue-sr-serde/blob/main/src/main/java/com/provectus/kafka/ui/serdes/glue/JsonUtil.java#L28)

But according to the comment in #3235 , that's not a completely valid fix since it can break other stuff?

Before and after the fix:
<img width="469" alt="Screen Shot 2023-01-15 at 15 48 52" src="https://user-images.githubusercontent.com/21235163/212545189-9b1e6b3f-d8b1-483a-a204-269150df8a93.png">
<img width="431" alt="Screen Shot 2023-01-15 at 15 45 44" src="https://user-images.githubusercontent.com/21235163/212545187-7eea90ee-f4c8-4a73-9dc9-17bfa4f0d86a.png">


Issue comments:
Hey, thanks, we'll take a look.
This seems to also apply to the default SchemaRegistry Serde. 

Is there a way i can fix this for the included SchemaRegistry serde?
@Haarolean Any update on this? We're also being affected by this odd display behavior and would like to see it fixed.
@frankgrimes97 planned for 0.8
Reopening, since it is only fixed for Kafka schema registry, not glue
@Haarolean Any update on when we might see a fix and 0.8 release?
> @Haarolean Any update on when we might see a fix and 0.8 release?

@frankgrimes97
https://github.com/provectus/kafka-ui/discussions/4255
https://github.com/kafbat/kafka-ui/discussions/23

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/JsonAvroConversionException.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/JsonAvroConversionException.java
new file mode 100644
index 00000000000..ef658031e56
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/JsonAvroConversionException.java
@@ -0,0 +1,7 @@
+package com.provectus.kafka.ui.exception;
+
+public class JsonAvroConversionException extends ValidationException {
+  public JsonAvroConversionException(String message) {
+    super(message);
+  }
+}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/JsonToAvroConversionException.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/JsonToAvroConversionException.java
deleted file mode 100644
index 5b3910f4cf3..00000000000
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/JsonToAvroConversionException.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package com.provectus.kafka.ui.exception;
-
-public class JsonToAvroConversionException extends ValidationException {
-  public JsonToAvroConversionException(String message) {
-    super(message);
-  }
-}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/AvroJsonSchemaConverter.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/AvroJsonSchemaConverter.java
index 56737d7a612..84f56b81dc0 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/AvroJsonSchemaConverter.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/AvroJsonSchemaConverter.java
@@ -151,30 +151,16 @@ private ArrayFieldSchema createArraySchema(Schema schema,
   }
 
   private JsonType convertType(Schema schema) {
-    switch (schema.getType()) {
-      case INT:
-      case LONG:
-        return new SimpleJsonType(JsonType.Type.INTEGER);
-      case MAP:
-      case RECORD:
-        return new SimpleJsonType(JsonType.Type.OBJECT);
-      case ENUM:
-        return new EnumJsonType(schema.getEnumSymbols());
-      case BYTES:
-      case STRING:
-        return new SimpleJsonType(JsonType.Type.STRING);
-      case NULL:
-        return new SimpleJsonType(JsonType.Type.NULL);
-      case ARRAY:
-        return new SimpleJsonType(JsonType.Type.ARRAY);
-      case FIXED:
-      case FLOAT:
-      case DOUBLE:
-        return new SimpleJsonType(JsonType.Type.NUMBER);
-      case BOOLEAN:
-        return new SimpleJsonType(JsonType.Type.BOOLEAN);
-      default:
-        return new SimpleJsonType(JsonType.Type.STRING);
-    }
+    return switch (schema.getType()) {
+      case INT, LONG -> new SimpleJsonType(JsonType.Type.INTEGER);
+      case MAP, RECORD -> new SimpleJsonType(JsonType.Type.OBJECT);
+      case ENUM -> new EnumJsonType(schema.getEnumSymbols());
+      case BYTES, STRING -> new SimpleJsonType(JsonType.Type.STRING);
+      case NULL -> new SimpleJsonType(JsonType.Type.NULL);
+      case ARRAY -> new SimpleJsonType(JsonType.Type.ARRAY);
+      case FIXED, FLOAT, DOUBLE -> new SimpleJsonType(JsonType.Type.NUMBER);
+      case BOOLEAN -> new SimpleJsonType(JsonType.Type.BOOLEAN);
+      default -> new SimpleJsonType(JsonType.Type.STRING);
+    };
   }
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversion.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversion.java
index d2114dd971c..0708044837b 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversion.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversion.java
@@ -1,6 +1,7 @@
 package com.provectus.kafka.ui.util.jsonschema;
 
 import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.json.JsonMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
@@ -15,7 +16,7 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.node.TextNode;
 import com.google.common.collect.Lists;
-import com.provectus.kafka.ui.exception.JsonToAvroConversionException;
+import com.provectus.kafka.ui.exception.JsonAvroConversionException;
 import io.confluent.kafka.serializers.AvroData;
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
@@ -34,7 +35,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.stream.Stream;
-import lombok.SneakyThrows;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 
@@ -42,12 +42,17 @@
 public class JsonAvroConversion {
 
   private static final JsonMapper MAPPER = new JsonMapper();
+  private static final Schema NULL_SCHEMA = Schema.create(Schema.Type.NULL);
 
   // converts json into Object that is expected input for KafkaAvroSerializer
   // (with AVRO_USE_LOGICAL_TYPE_CONVERTERS flat enabled!)
-  @SneakyThrows
   public static Object convertJsonToAvro(String jsonString, Schema avroSchema) {
-    JsonNode rootNode = MAPPER.readTree(jsonString);
+    JsonNode rootNode = null;
+    try {
+      rootNode = MAPPER.readTree(jsonString);
+    } catch (JsonProcessingException e) {
+      throw new JsonAvroConversionException("String is not a valid json");
+    }
     return convert(rootNode, avroSchema);
   }
 
@@ -80,7 +85,7 @@ private static Object convert(JsonNode node, Schema avroSchema) {
         assertJsonType(node, JsonNodeType.STRING);
         String symbol = node.textValue();
         if (!avroSchema.getEnumSymbols().contains(symbol)) {
-          throw new JsonToAvroConversionException("%s is not a part of enum symbols [%s]"
+          throw new JsonAvroConversionException("%s is not a part of enum symbols [%s]"
               .formatted(symbol, avroSchema.getEnumSymbols()));
         }
         yield new GenericData.EnumSymbol(avroSchema, symbol);
@@ -88,23 +93,35 @@ private static Object convert(JsonNode node, Schema avroSchema) {
       case UNION -> {
         // for types from enum (other than null) payload should be an object with single key == name of type
         // ex: schema = [ "null", "int", "string" ], possible payloads = null, { "string": "str" },  { "int": 123 }
-        if (node.isNull() && avroSchema.getTypes().contains(Schema.create(Schema.Type.NULL))) {
+        if (node.isNull() && avroSchema.getTypes().contains(NULL_SCHEMA)) {
           yield null;
         }
 
         assertJsonType(node, JsonNodeType.OBJECT);
         var elements = Lists.newArrayList(node.fields());
         if (elements.size() != 1) {
-          throw new JsonToAvroConversionException(
+          throw new JsonAvroConversionException(
               "UNION field value should be an object with single field == type name");
         }
-        var typeNameToValue = elements.get(0);
+        Map.Entry<String, JsonNode> typeNameToValue = elements.get(0);
+        List<Schema> candidates = new ArrayList<>();
         for (Schema unionType : avroSchema.getTypes()) {
           if (typeNameToValue.getKey().equals(unionType.getFullName())) {
             yield convert(typeNameToValue.getValue(), unionType);
           }
+          if (typeNameToValue.getKey().equals(unionType.getName())) {
+            candidates.add(unionType);
+          }
+        }
+        if (candidates.size() == 1) {
+          yield convert(typeNameToValue.getValue(), candidates.get(0));
         }
-        throw new JsonToAvroConversionException(
+        if (candidates.size() > 1) {
+          throw new JsonAvroConversionException(
+              "Can't select type within union for value '%s'. Provide full type name.".formatted(node)
+          );
+        }
+        throw new JsonAvroConversionException(
             "json value '%s' is cannot be converted to any of union types [%s]"
                 .formatted(node, avroSchema.getTypes()));
       }
@@ -164,7 +181,7 @@ private static Object convert(JsonNode node, Schema avroSchema) {
         assertJsonType(node, JsonNodeType.STRING);
         byte[] bytes = node.textValue().getBytes(StandardCharsets.ISO_8859_1);
         if (bytes.length != avroSchema.getFixedSize()) {
-          throw new JsonToAvroConversionException(
+          throw new JsonAvroConversionException(
               "Fixed field has unexpected size %d (should be %d)"
                   .formatted(bytes.length, avroSchema.getFixedSize()));
         }
@@ -208,8 +225,11 @@ public static JsonNode convertAvroToJson(Object obj, Schema avroSchema) {
       case UNION -> {
         ObjectNode node = MAPPER.createObjectNode();
         int unionIdx = AvroData.getGenericData().resolveUnion(avroSchema, obj);
-        Schema unionType = avroSchema.getTypes().get(unionIdx);
-        node.set(unionType.getFullName(), convertAvroToJson(obj, unionType));
+        Schema selectedType = avroSchema.getTypes().get(unionIdx);
+        node.set(
+            selectUnionTypeFieldName(avroSchema, selectedType, unionIdx),
+            convertAvroToJson(obj, selectedType)
+        );
         yield node;
       }
       case STRING -> {
@@ -252,11 +272,30 @@ public static JsonNode convertAvroToJson(Object obj, Schema avroSchema) {
     };
   }
 
+  // select name for a key field that represents type name of union.
+  // For records selects short name, if it is possible.
+  private static String selectUnionTypeFieldName(Schema unionSchema,
+                                                 Schema chosenType,
+                                                 int chosenTypeIdx) {
+    var types = unionSchema.getTypes();
+    if (types.size() == 2 && types.contains(NULL_SCHEMA)) {
+      return chosenType.getName();
+    }
+    for (int i = 0; i < types.size(); i++) {
+      if (i != chosenTypeIdx && chosenType.getName().equals(types.get(i).getName())) {
+        // there is another type inside union with the same name
+        // so, we have to use fullname
+        return chosenType.getFullName();
+      }
+    }
+    return chosenType.getName();
+  }
+
   private static Object processLogicalType(JsonNode node, Schema schema) {
     return findConversion(schema)
         .map(c -> c.jsonToAvroConversion.apply(node, schema))
         .orElseThrow(() ->
-            new JsonToAvroConversionException("'%s' logical type is not supported"
+            new JsonAvroConversionException("'%s' logical type is not supported"
                 .formatted(schema.getLogicalType().getName())));
   }
 
@@ -264,7 +303,7 @@ private static JsonNode processLogicalType(Object obj, Schema schema) {
     return findConversion(schema)
         .map(c -> c.avroToJsonConversion.apply(obj, schema))
         .orElseThrow(() ->
-            new JsonToAvroConversionException("'%s' logical type is not supported"
+            new JsonAvroConversionException("'%s' logical type is not supported"
                 .formatted(schema.getLogicalType().getName())));
   }
 
@@ -281,7 +320,7 @@ private static boolean isLogicalType(Schema schema) {
 
   private static void assertJsonType(JsonNode node, JsonNodeType... allowedTypes) {
     if (Stream.of(allowedTypes).noneMatch(t -> node.getNodeType() == t)) {
-      throw new JsonToAvroConversionException(
+      throw new JsonAvroConversionException(
           "%s node has unexpected type, allowed types %s, actual type %s"
               .formatted(node, Arrays.toString(allowedTypes), node.getNodeType()));
     }
@@ -289,7 +328,7 @@ private static void assertJsonType(JsonNode node, JsonNodeType... allowedTypes)
 
   private static void assertJsonNumberType(JsonNode node, JsonParser.NumberType... allowedTypes) {
     if (Stream.of(allowedTypes).noneMatch(t -> node.numberType() == t)) {
-      throw new JsonToAvroConversionException(
+      throw new JsonAvroConversionException(
           "%s node has unexpected numeric type, allowed types %s, actual type %s"
               .formatted(node, Arrays.toString(allowedTypes), node.numberType()));
     }
@@ -318,7 +357,7 @@ enum LogicalTypeConversion {
           } else if (node.isNumber()) {
             return new BigDecimal(node.numberValue().toString());
           }
-          throw new JsonToAvroConversionException(
+          throw new JsonAvroConversionException(
               "node '%s' can't be converted to decimal logical type"
                   .formatted(node));
         },
@@ -335,7 +374,7 @@ enum LogicalTypeConversion {
           } else if (node.isTextual()) {
             return LocalDate.parse(node.asText());
           } else {
-            throw new JsonToAvroConversionException(
+            throw new JsonAvroConversionException(
                 "node '%s' can't be converted to date logical type"
                     .formatted(node));
           }
@@ -356,7 +395,7 @@ enum LogicalTypeConversion {
           } else if (node.isTextual()) {
             return LocalTime.parse(node.asText());
           } else {
-            throw new JsonToAvroConversionException(
+            throw new JsonAvroConversionException(
                 "node '%s' can't be converted to time-millis logical type"
                     .formatted(node));
           }
@@ -377,7 +416,7 @@ enum LogicalTypeConversion {
           } else if (node.isTextual()) {
             return LocalTime.parse(node.asText());
           } else {
-            throw new JsonToAvroConversionException(
+            throw new JsonAvroConversionException(
                 "node '%s' can't be converted to time-micros logical type"
                     .formatted(node));
           }
@@ -398,7 +437,7 @@ enum LogicalTypeConversion {
           } else if (node.isTextual()) {
             return Instant.parse(node.asText());
           } else {
-            throw new JsonToAvroConversionException(
+            throw new JsonAvroConversionException(
                 "node '%s' can't be converted to timestamp-millis logical type"
                     .formatted(node));
           }
@@ -423,7 +462,7 @@ enum LogicalTypeConversion {
           } else if (node.isTextual()) {
             return Instant.parse(node.asText());
           } else {
-            throw new JsonToAvroConversionException(
+            throw new JsonAvroConversionException(
                 "node '%s' can't be converted to timestamp-millis logical type"
                     .formatted(node));
           }


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversionTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversionTest.java
index 0e9c291707e..7c4d79f30d7 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversionTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/jsonschema/JsonAvroConversionTest.java
@@ -3,6 +3,7 @@
 import static com.provectus.kafka.ui.util.jsonschema.JsonAvroConversion.convertAvroToJson;
 import static com.provectus.kafka.ui.util.jsonschema.JsonAvroConversion.convertJsonToAvro;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.json.JsonMapper;
@@ -13,6 +14,7 @@
 import com.fasterxml.jackson.databind.node.LongNode;
 import com.fasterxml.jackson.databind.node.TextNode;
 import com.google.common.primitives.Longs;
+import com.provectus.kafka.ui.exception.JsonAvroConversionException;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
@@ -181,12 +183,62 @@ var record = (GenericData.Record) convertJsonToAvro(jsonPayload, schema);
       record = (GenericData.Record) convertJsonToAvro(jsonPayload, schema);
       assertThat(record.get("f_union")).isEqualTo(123);
 
-      //inner-record's name should be fully-qualified!
-      jsonPayload = "{ \"f_union\": { \"com.test.TestAvroRecord\": { \"f_union\": { \"int\": 123  } } } }";
+      //short name can be used since there is no clash with other type names
+      jsonPayload = "{ \"f_union\": { \"TestAvroRecord\": { \"f_union\": { \"int\": 123  } } } }";
       record = (GenericData.Record) convertJsonToAvro(jsonPayload, schema);
       assertThat(record.get("f_union")).isInstanceOf(GenericData.Record.class);
       var innerRec = (GenericData.Record) record.get("f_union");
       assertThat(innerRec.get("f_union")).isEqualTo(123);
+
+      assertThatThrownBy(() ->
+          convertJsonToAvro("{ \"f_union\": { \"NotExistingType\": 123 } }", schema)
+      ).isInstanceOf(JsonAvroConversionException.class);
+    }
+
+    @Test
+    void unionFieldWithTypeNamesClash() {
+      var schema = createSchema(
+          """
+               {
+                 "type": "record",
+                 "namespace": "com.test",
+                 "name": "TestAvroRecord",
+                 "fields": [
+                   {
+                     "name": "nestedClass",
+                     "type": {
+                       "type": "record",
+                       "namespace": "com.nested",
+                       "name": "TestAvroRecord",
+                       "fields": [
+                         {"name" : "inner_obj_field", "type": "int" }
+                       ]
+                     }
+                   },
+                   {
+                     "name": "f_union",
+                     "type": [ "null", "int", "com.test.TestAvroRecord", "com.nested.TestAvroRecord"]
+                   }
+                 ]
+              }"""
+      );
+      //short name can't can be used since there is a clash with other type names
+      var jsonPayload = "{ \"f_union\": { \"com.test.TestAvroRecord\": { \"f_union\": { \"int\": 123  } } } }";
+      var record = (GenericData.Record) convertJsonToAvro(jsonPayload, schema);
+      assertThat(record.get("f_union")).isInstanceOf(GenericData.Record.class);
+      var innerRec = (GenericData.Record) record.get("f_union");
+      assertThat(innerRec.get("f_union")).isEqualTo(123);
+
+      //short name can't can be used since there is a clash with other type names
+      jsonPayload = "{ \"f_union\": { \"com.nested.TestAvroRecord\": { \"inner_obj_field\":  234 } } }";
+      record = (GenericData.Record) convertJsonToAvro(jsonPayload, schema);
+      assertThat(record.get("f_union")).isInstanceOf(GenericData.Record.class);
+      innerRec = (GenericData.Record) record.get("f_union");
+      assertThat(innerRec.get("inner_obj_field")).isEqualTo(234);
+
+      assertThatThrownBy(() ->
+          convertJsonToAvro("{ \"f_union\": { \"TestAvroRecord\": { \"inner_obj_field\":  234 } } }", schema)
+      ).isInstanceOf(JsonAvroConversionException.class);
     }
 
     @Test
@@ -599,6 +651,46 @@ void unionField() {
       var innerRec = new GenericData.Record(schema);
       innerRec.put("f_union", 123);
       r.put("f_union", innerRec);
+      // short type name can be set since there is NO clash with other types name
+      assertJsonsEqual(
+          " { \"f_union\" : { \"TestAvroRecord\" : { \"f_union\" : { \"int\" : 123 } } } }",
+          convertAvroToJson(r, schema)
+      );
+    }
+
+    @Test
+    void unionFieldWithInnerTypesNamesClash() {
+      var schema = createSchema(
+          """
+               {
+                 "type": "record",
+                 "namespace": "com.test",
+                 "name": "TestAvroRecord",
+                 "fields": [
+                   {
+                     "name": "nestedClass",
+                     "type": {
+                       "type": "record",
+                       "namespace": "com.nested",
+                       "name": "TestAvroRecord",
+                       "fields": [
+                         {"name" : "inner_obj_field", "type": "int" }
+                       ]
+                     }
+                   },
+                   {
+                     "name": "f_union",
+                     "type": [ "null", "int", "com.test.TestAvroRecord", "com.nested.TestAvroRecord"]
+                   }
+                 ]
+              }"""
+      );
+
+      var r = new GenericData.Record(schema);
+      var innerRec = new GenericData.Record(schema);
+      innerRec.put("f_union", 123);
+      r.put("f_union", innerRec);
+      // full type name should be set since there is a clash with other type name
       assertJsonsEqual(
           " { \"f_union\" : { \"com.test.TestAvroRecord\" : { \"f_union\" : { \"int\" : 123 } } } }",
           convertAvroToJson(r, schema)