Fix decimal serialization

src/main/java/org/akhq/models/Record.java

@@ -3,12 +3,12 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 import lombok.*;
+import org.akhq.utils.AvroToJsonSerializer;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.record.TimestampType;
-import org.joda.time.DateTime;
 
 import java.nio.ByteBuffer;
 import java.time.Instant;
@@ -107,8 +107,8 @@ private String convertToString(byte[] payload, Integer keySchemaId) {
             return null;
         } else  if (keySchemaId != null) {
             try {
-                GenericRecord deserialize = (GenericRecord) kafkaAvroDeserializer.deserialize(topic, payload);
-                return deserialize.toString();
+                GenericRecord record = (GenericRecord) kafkaAvroDeserializer.deserialize(topic, payload);
+                return AvroToJsonSerializer.toJson(record);
             } catch (Exception exception) {
                 return new String(payload);
             }

src/main/java/org/akhq/models/Schema.java

@@ -5,8 +5,8 @@
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.google.common.annotations.VisibleForTesting;
 import lombok.*;
-import org.akhq.utils.AvroDeserializer;
-import org.akhq.utils.AvroSerializer;
+import org.akhq.utils.AvroSchemaDeserializer;
+import org.akhq.utils.AvroSchemaSerializer;
 import org.apache.avro.AvroTypeException;
 import org.apache.avro.Schema.Parser;
 
@@ -28,8 +28,8 @@ public class Schema {
     private Integer version;
     private Config.CompatibilityLevelConfig compatibilityLevel;
 
-    @JsonSerialize(using = AvroSerializer.class)
-    @JsonDeserialize(using = AvroDeserializer.class)
+    @JsonSerialize(using = AvroSchemaSerializer.class)
+    @JsonDeserialize(using = AvroSchemaDeserializer.class)
     private org.apache.avro.Schema schema;
 
     private String exception;

src/main/java/org/akhq/utils/AvroDeserializer.java → src/main/java/org/akhq/utils/AvroSchemaDeserializer.java

@@ -8,7 +8,7 @@
 
 import java.io.IOException;
 
-public class AvroDeserializer extends JsonDeserializer<Schema> {
+public class AvroSchemaDeserializer extends JsonDeserializer<Schema> {
     @Override
     public Schema deserialize(
         JsonParser p,

src/main/java/org/akhq/utils/AvroSerializer.java → src/main/java/org/akhq/utils/AvroSchemaSerializer.java

@@ -8,7 +8,7 @@
 
 import java.io.IOException;
 
-public class AvroSerializer extends JsonSerializer<Schema> {
+public class AvroSchemaSerializer extends JsonSerializer<Schema> {
     @Override
     public void serialize(
         Schema value,

src/main/java/org/akhq/utils/AvroToJsonSerializer.java

@@ -0,0 +1,63 @@
+package org.akhq.utils;
+
+import org.apache.avro.*;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.Encoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.avro.io.JsonEncoder;
+import org.apache.avro.specific.SpecificDatumWriter;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+
+public class AvroToJsonSerializer {
+
+    public static String toJson(GenericRecord record) throws IOException {
+        try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
+            JsonEncoder jsonEncoder = EncoderFactory.get().jsonEncoder(record.getSchema(), outputStream);
+            new BigDecimalFriendlySpecificDatumWriter<GenericRecord>(record.getSchema()).write(record, jsonEncoder);
+            jsonEncoder.flush();
+            return new String(outputStream.toByteArray());
+        }
+    }
+}
+
+class BigDecimalFriendlySpecificDatumWriter<T> extends SpecificDatumWriter<T> {
+
+    private static final Conversion<BigDecimal> DECIMAL_CONVERSION = new Conversions.DecimalConversion();
+
+    public BigDecimalFriendlySpecificDatumWriter(Schema schema) {
+        super(schema);
+    }
+
+    @Override
+    protected void writeField(Object datum, Schema.Field f, Encoder out, Object state) throws IOException {
+        if (datum instanceof GenericData.Record) {
+            Schema fieldSchema = f.schema();
+            LogicalType logicalType = fieldSchema.getLogicalType();
+            Object value = getData().getField(datum, f.name(), f.pos());
+            if (logicalType instanceof LogicalTypes.Decimal) {
+                value = convert(DECIMAL_CONVERSION, fieldSchema, logicalType, value);
+            }
+            writeWithoutConversion(fieldSchema, value, out);
+        } else {
+            super.writeField(datum, f, out, state);
+        }
+    }
+
+    private Object convert(Conversion<?> conversion, Schema fieldSchema, LogicalType logicalType, Object value) {
+        if (conversion instanceof Conversions.DecimalConversion && value instanceof ByteBuffer) {
+            // convert decimal value to a string
+            byte[] byteValue = new byte[((ByteBuffer) value).remaining()];
+            ((ByteBuffer) value).get(byteValue);
+            BigDecimal number = (BigDecimal) conversion.fromBytes(ByteBuffer.wrap(byteValue), fieldSchema, logicalType);
+            return (ByteBuffer.wrap(number.toPlainString().getBytes()));
+        } else {
+            return convert(fieldSchema, logicalType, conversion, value);
+        }
+    }
+}
+

src/test/avro/Dog.avsc

@@ -0,0 +1,10 @@
+{
+    "name": "Dog",
+    "namespace": "org.akhq",
+    "type": "record",
+    "fields" : [
+        {"name": "id", "type": "int"},
+        {"name": "name", "type": "string"},
+        {"name": "weight", "type" : { "type" : "bytes", "logicalType" : "decimal", "precision" : 4, "scale" : 2}}
+    ]
+}

src/test/java/org/akhq/modules/AvroSerializerTest.java → src/test/java/org/akhq/modules/AvroSchemaSerializerTest.java

@@ -19,7 +19,7 @@
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
-class AvroSerializerTest {
+class AvroSchemaSerializerTest {
 
     private final org.apache.avro.Schema SCHEMA = SchemaBuilder
             .record("schema1").namespace("org.akhq")

src/test/java/org/akhq/modules/AvroToJsonSerializerTest.java

@@ -0,0 +1,54 @@
+package org.akhq.modules;
+
+import org.akhq.Breed;
+import org.akhq.Cat;
+import org.akhq.Dog;
+import org.akhq.utils.AvroToJsonSerializer;
+import org.apache.avro.Conversion;
+import org.apache.avro.Conversions;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class AvroToJsonSerializerTest {
+
+    private static final Conversion<BigDecimal> DECIMAL_CONVERSION = new Conversions.DecimalConversion();
+
+    @Test
+    public void serializeAvroToJsonWithDecimal() throws IOException {
+        String expectedString = "{\"id\":10,\"name\":\"Tiger\",\"weight\":\"10.40\"}";
+        GenericRecord dogExample = aDogExample(10, "Tiger", 10.40);
+        assertEquals(AvroToJsonSerializer.toJson(dogExample), expectedString);
+    }
+
+    @Test
+    public void serializeAvroToJsonWithoutDecimal() throws IOException {
+        String expectedString = "{\"id\":10,\"name\":\"Tom\",\"breed\":\"SPHYNX\"}";
+        GenericRecord catExample = aCatExample(10, "Tom", Breed.SPHYNX);
+        assertEquals(AvroToJsonSerializer.toJson(catExample), expectedString);
+    }
+
+    private GenericRecord aCatExample(int id, String name, Breed breed) {
+        return new GenericRecordBuilder(Cat.SCHEMA$)
+                .set("id", id)
+                .set("name", name)
+                .set("breed", breed)
+                .build();
+    }
+
+    private GenericRecord aDogExample(int id, String name, double weight) {
+        Schema.Field weightField = Dog.SCHEMA$.getField("weight");
+        return new GenericRecordBuilder(Dog.SCHEMA$)
+                .set("id", id)
+                .set("name", name)
+                .set("weight", DECIMAL_CONVERSION.toBytes(BigDecimal.valueOf(weight).setScale(2), weightField.schema(), weightField.schema().getLogicalType()))
+                .build();
+    }
+
+}

src/test/java/org/akhq/repositories/RecordRepositoryTest.java

@@ -122,7 +122,7 @@ public void consumeAvro() throws ExecutionException, InterruptedException {
             .findFirst();
 
         avroRecord.orElseThrow(() -> new NoSuchElementException("Unable to find key 1"));
-        avroRecord.ifPresent(record -> assertEquals("{\"id\": 1, \"name\": \"WaWa\", \"breed\": \"ABYSSINIAN\"}", record.getValue()));
+        avroRecord.ifPresent(record -> assertEquals("{\"id\":1,\"name\":\"WaWa\",\"breed\":\"ABYSSINIAN\"}", record.getValue()));
     }
 
     private List<Record> consumeAllRecord(RecordRepository.Options options) throws ExecutionException, InterruptedException {