[ESQUE-21] Add ExtendedJsonDecoder for more convenient AVRO publishing

patschuh · Mar 30, 2019 · 5e20203 · 5e20203
1 parent eb6f9bb
commit 5e20203
Show file tree

Hide file tree

Showing 2 changed files with 786 additions and 3 deletions.
diff --git a/src/main/java/at/esque/kafka/handlers/ProducerHandler.java b/src/main/java/at/esque/kafka/handlers/ProducerHandler.java
@@ -3,6 +3,7 @@
 import at.esque.kafka.MessageType;
 import at.esque.kafka.cluster.ClusterConfig;
 import at.esque.kafka.cluster.TopicMessageTypeConfig;
+import at.esque.kafka.serialization.ExtendedJsonDecoder;
 import at.esque.kafka.serialization.KafkaEsqueSerializer;
 import com.google.inject.Inject;
 import io.confluent.kafka.schemaregistry.client.rest.RestService;
@@ -11,8 +12,7 @@
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.DatumReader;
-import org.apache.avro.io.DecoderFactory;
-import org.apache.avro.io.JsonDecoder;
+import org.apache.avro.io.Decoder;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -128,7 +128,7 @@ private GenericRecord createRecord(ProducerWrapper producerWrapper, String json,
 
         org.apache.avro.Schema avroSchema = new org.apache.avro.Schema.Parser().parse(schema.getSchema());
 
-        JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(avroSchema, json);
+        Decoder jsonDecoder = new ExtendedJsonDecoder(avroSchema, json);
         DatumReader<GenericRecord> reader = new GenericDatumReader<>(avroSchema);
 
         return reader.read(null, jsonDecoder);