Unnecessary casting removed from AvroFormatter to fix primitive values deserialization (#902)

* Unnecessary casting removed from AvroFormatter to prevent primitive values deserialization

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

@@ -4,7 +4,6 @@ import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 import lombok.SneakyThrows;
-import org.apache.avro.generic.GenericRecord;
 
 public class AvroMessageFormatter implements MessageFormatter {
   private final KafkaAvroDeserializer avroDeserializer;
@@ -16,8 +15,10 @@ public class AvroMessageFormatter implements MessageFormatter {
   @Override
   @SneakyThrows
   public String format(String topic, byte[] value) {
-    GenericRecord avroRecord = (GenericRecord) avroDeserializer.deserialize(topic, value);
-    byte[] jsonBytes = AvroSchemaUtils.toJson(avroRecord);
+    // deserialized will have type, that depends on schema type (record or primitive),
+    // AvroSchemaUtils.toJson(...) method will take it into account
+    Object deserialized = avroDeserializer.deserialize(topic, value);
+    byte[] jsonBytes = AvroSchemaUtils.toJson(deserialized);
     return new String(jsonBytes);
   }
 

kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/SendAndReadTests.java

@@ -65,6 +65,13 @@ public class SendAndReadTests extends AbstractBaseTest {
           + "}"
   );
 
+  private static final AvroSchema AVRO_SCHEMA_PRIMITIVE_STRING =
+      new AvroSchema("{ \"type\": \"string\" }");
+
+  private static final AvroSchema AVRO_SCHEMA_PRIMITIVE_INT =
+      new AvroSchema("{ \"type\": \"int\" }");
+
+
   private static final String AVRO_SCHEMA_1_JSON_RECORD
       = "{ \"field1\":\"testStr\", \"field2\": 123 }";
 
@@ -187,6 +194,22 @@ public class SendAndReadTests extends AbstractBaseTest {
         });
   }
 
+  @Test
+  void primitiveAvroSchemas() {
+    new SendAndReadSpec()
+        .withKeySchema(AVRO_SCHEMA_PRIMITIVE_STRING)
+        .withValueSchema(AVRO_SCHEMA_PRIMITIVE_INT)
+        .withMsgToSend(
+            new CreateTopicMessage()
+                .key("\"some string\"")
+                .content("123")
+        )
+        .doAssert(polled -> {
+          assertThat(polled.getKey()).isEqualTo("\"some string\"");
+          assertThat(polled.getContent()).isEqualTo("123");
+        });
+  }
+
   @Test
   void nonNullableKvWithAvroSchema() {
     new SendAndReadSpec()