diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/AvroMessageFormatter.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/AvroMessageFormatter.java index 817ade0fb51..e69d522ea03 100644 --- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/AvroMessageFormatter.java +++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/AvroMessageFormatter.java @@ -4,7 +4,6 @@ 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 AvroMessageFormatter(SchemaRegistryClient client) { @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); } diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/SendAndReadTests.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/SendAndReadTests.java index 7bd16af2725..3b578a6a892 100644 --- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/SendAndReadTests.java +++ b/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 @@ void noSchemaValueIsNull() { }); } + @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()