Index: src/main/java/at/esque/kafka/handlers/ProducerHandler.java IDEA additional info: Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP <+>UTF-8 =================================================================== --- src/main/java/at/esque/kafka/handlers/ProducerHandler.java (revision fd97d0bbeae497baa1fcc16818c7ff5af690b484) +++ src/main/java/at/esque/kafka/handlers/ProducerHandler.java (date 1553095428179) @@ -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; @@ -124,7 +124,7 @@ 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 reader = new GenericDatumReader<>(avroSchema); return reader.read(null, jsonDecoder); Index: src/main/java/at/esque/kafka/serialization/ExtendedJsonDecoder.java IDEA additional info: Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP <+>UTF-8 =================================================================== --- src/main/java/at/esque/kafka/serialization/ExtendedJsonDecoder.java (date 1553095428169) +++ src/main/java/at/esque/kafka/serialization/ExtendedJsonDecoder.java (date 1553095428169) @@ -0,0 +1,783 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package at.esque.kafka.serialization; + +import org.apache.avro.AvroTypeException; +import org.apache.avro.Schema; +import org.apache.avro.Schema.Field; +import org.apache.avro.Schema.Type; +import org.apache.avro.io.Decoder; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.io.ParsingDecoder; +import org.apache.avro.io.parsing.JsonGrammarGenerator; +import org.apache.avro.io.parsing.Parser; +import org.apache.avro.io.parsing.Symbol; +import org.apache.avro.util.Utf8; +import org.codehaus.jackson.Base64Variant; +import org.codehaus.jackson.JsonFactory; +import org.codehaus.jackson.JsonLocation; +import org.codehaus.jackson.JsonNode; +import org.codehaus.jackson.JsonParser; +import org.codehaus.jackson.JsonStreamContext; +import org.codehaus.jackson.JsonToken; +import org.codehaus.jackson.ObjectCodec; +import org.codehaus.jackson.node.NullNode; + +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Stack; + +/** + * Original code is from: https://github.com/Celos/avro-json-decoder + * + * Additional change for KafkaEsque: a NullNode.getInstance() is returned for fields that are missing from the JSON, + * meaning they are not explicitly set to null in the JSON. + * + * A {@link Decoder} for Avro's JSON data encoding. + *

+ * Construct using {@link DecoderFactory}. + *

+ * ExtendedJsonDecoder is not thread-safe. + *

+ * Based on {@link org.apache.avro.io.JsonDecoder JsonDecoder} + * and ExtendedJsonDecoder. + * Infers default arguments, if they are not present. + *

+ **/ +public class ExtendedJsonDecoder extends ParsingDecoder + implements Parser.ActionHandler { + private JsonParser in; + private static JsonFactory jsonFactory = new JsonFactory(); + Stack reorderBuffers = new Stack(); + ReorderBuffer currentReorderBuffer; + + private final Schema schema; + + private static class ReorderBuffer { + public Map> savedFields = new HashMap>(); + public JsonParser origParser = null; + } + + static final String CHARSET = "ISO-8859-1"; + + public ExtendedJsonDecoder(Schema schema, InputStream in) throws IOException { + super(getSymbol(schema)); + configure(in); + this.schema = schema; + } + + public ExtendedJsonDecoder(Schema schema, String in) throws IOException { + super(getSymbol(schema)); + configure(in); + this.schema = schema; + } + + private static Symbol getSymbol(Schema schema) { + if (null == schema) { + throw new NullPointerException("Schema cannot be null!"); + } + return new JsonGrammarGenerator().generate(schema); + } + + /** + * Reconfigures this JsonDecoder to use the InputStream provided. + *

+ * If the InputStream provided is null, a NullPointerException is thrown. + *

+ * Otherwise, this JsonDecoder will reset its state and then + * reconfigure its input. + * @param in + * The IntputStream to read from. Cannot be null. + * @throws IOException + * @return this JsonDecoder + */ + public ExtendedJsonDecoder configure(InputStream in) throws IOException { + if (null == in) { + throw new NullPointerException("InputStream to read from cannot be null!"); + } + parser.reset(); + this.in = jsonFactory.createJsonParser(in); + this.in.nextToken(); + return this; + } + + /** + * Reconfigures this JsonDecoder to use the String provided for input. + *

+ * If the String provided is null, a NullPointerException is thrown. + *

+ * Otherwise, this JsonDecoder will reset its state and then + * reconfigure its input. + * @param in + * The String to read from. Cannot be null. + * @throws IOException + * @return this JsonDecoder + */ + public ExtendedJsonDecoder configure(String in) throws IOException { + if (null == in) { + throw new NullPointerException("String to read from cannot be null!"); + } + parser.reset(); + this.in = new JsonFactory().createJsonParser(in); + this.in.nextToken(); + return this; + } + + private void advance(Symbol symbol) throws IOException { + this.parser.processTrailingImplicitActions(); + if (in.getCurrentToken() == null && this.parser.depth() == 1) + throw new EOFException(); + parser.advance(symbol); + } + + @Override + public void readNull() throws IOException { + advance(Symbol.NULL); + if (in.getCurrentToken() == JsonToken.VALUE_NULL) { + in.nextToken(); + } else { + throw error("null"); + } + } + + @Override + public boolean readBoolean() throws IOException { + advance(Symbol.BOOLEAN); + JsonToken t = in.getCurrentToken(); + if (t == JsonToken.VALUE_TRUE || t == JsonToken.VALUE_FALSE) { + in.nextToken(); + return t == JsonToken.VALUE_TRUE; + } else { + throw error("boolean"); + } + } + + @Override + public int readInt() throws IOException { + advance(Symbol.INT); + if (in.getCurrentToken().isNumeric()) { + int result = in.getIntValue(); + in.nextToken(); + return result; + } else { + throw error("int"); + } + } + + @Override + public long readLong() throws IOException { + advance(Symbol.LONG); + if (in.getCurrentToken().isNumeric()) { + long result = in.getLongValue(); + in.nextToken(); + return result; + } else { + throw error("long"); + } + } + + @Override + public float readFloat() throws IOException { + advance(Symbol.FLOAT); + if (in.getCurrentToken().isNumeric()) { + float result = in.getFloatValue(); + in.nextToken(); + return result; + } else { + throw error("float"); + } + } + + @Override + public double readDouble() throws IOException { + advance(Symbol.DOUBLE); + if (in.getCurrentToken().isNumeric()) { + double result = in.getDoubleValue(); + in.nextToken(); + return result; + } else { + throw error("double"); + } + } + + @Override + public Utf8 readString(Utf8 old) throws IOException { + return new Utf8(readString()); + } + + @Override + public String readString() throws IOException { + advance(Symbol.STRING); + if (parser.topSymbol() == Symbol.MAP_KEY_MARKER) { + parser.advance(Symbol.MAP_KEY_MARKER); + if (in.getCurrentToken() != JsonToken.FIELD_NAME) { + throw error("map-key"); + } + } else { + if (in.getCurrentToken() != JsonToken.VALUE_STRING) { + throw error("string"); + } + } + String result = in.getText(); + in.nextToken(); + return result; + } + + @Override + public void skipString() throws IOException { + advance(Symbol.STRING); + if (parser.topSymbol() == Symbol.MAP_KEY_MARKER) { + parser.advance(Symbol.MAP_KEY_MARKER); + if (in.getCurrentToken() != JsonToken.FIELD_NAME) { + throw error("map-key"); + } + } else { + if (in.getCurrentToken() != JsonToken.VALUE_STRING) { + throw error("string"); + } + } + in.nextToken(); + } + + @Override + public ByteBuffer readBytes(ByteBuffer old) throws IOException { + advance(Symbol.BYTES); + if (in.getCurrentToken() == JsonToken.VALUE_STRING) { + byte[] result = readByteArray(); + in.nextToken(); + return ByteBuffer.wrap(result); + } else { + throw error("bytes"); + } + } + + private byte[] readByteArray() throws IOException { + byte[] result = in.getText().getBytes(CHARSET); + return result; + } + + @Override + public void skipBytes() throws IOException { + advance(Symbol.BYTES); + if (in.getCurrentToken() == JsonToken.VALUE_STRING) { + in.nextToken(); + } else { + throw error("bytes"); + } + } + + private void checkFixed(int size) throws IOException { + advance(Symbol.FIXED); + Symbol.IntCheckAction top = (Symbol.IntCheckAction) parser.popSymbol(); + if (size != top.size) { + throw new AvroTypeException( + "Incorrect length for fixed binary: expected " + + top.size + " but received " + size + " bytes."); + } + } + + @Override + public void readFixed(byte[] bytes, int start, int len) throws IOException { + checkFixed(len); + if (in.getCurrentToken() == JsonToken.VALUE_STRING) { + byte[] result = readByteArray(); + in.nextToken(); + if (result.length != len) { + throw new AvroTypeException("Expected fixed length " + len + + ", but got" + result.length); + } + System.arraycopy(result, 0, bytes, start, len); + } else { + throw error("fixed"); + } + } + + @Override + public void skipFixed(int length) throws IOException { + checkFixed(length); + doSkipFixed(length); + } + + private void doSkipFixed(int length) throws IOException { + if (in.getCurrentToken() == JsonToken.VALUE_STRING) { + byte[] result = readByteArray(); + in.nextToken(); + if (result.length != length) { + throw new AvroTypeException("Expected fixed length " + length + + ", but got" + result.length); + } + } else { + throw error("fixed"); + } + } + + @Override + protected void skipFixed() throws IOException { + advance(Symbol.FIXED); + Symbol.IntCheckAction top = (Symbol.IntCheckAction) parser.popSymbol(); + doSkipFixed(top.size); + } + + @Override + public int readEnum() throws IOException { + advance(Symbol.ENUM); + Symbol.EnumLabelsAction top = (Symbol.EnumLabelsAction) parser.popSymbol(); + if (in.getCurrentToken() == JsonToken.VALUE_STRING) { + in.getText(); + int n = top.findLabel(in.getText()); + if (n >= 0) { + in.nextToken(); + return n; + } + throw new AvroTypeException("Unknown symbol in enum " + in.getText()); + } else { + throw error("fixed"); + } + } + + @Override + public long readArrayStart() throws IOException { + advance(Symbol.ARRAY_START); + if (in.getCurrentToken() == JsonToken.START_ARRAY) { + in.nextToken(); + return doArrayNext(); + } else { + throw error("array-start"); + } + } + + @Override + public long arrayNext() throws IOException { + advance(Symbol.ITEM_END); + return doArrayNext(); + } + + private long doArrayNext() throws IOException { + if (in.getCurrentToken() == JsonToken.END_ARRAY) { + parser.advance(Symbol.ARRAY_END); + in.nextToken(); + return 0; + } else { + return 1; + } + } + + @Override + public long skipArray() throws IOException { + advance(Symbol.ARRAY_START); + if (in.getCurrentToken() == JsonToken.START_ARRAY) { + in.skipChildren(); + in.nextToken(); + advance(Symbol.ARRAY_END); + } else { + throw error("array-start"); + } + return 0; + } + + @Override + public long readMapStart() throws IOException { + advance(Symbol.MAP_START); + if (in.getCurrentToken() == JsonToken.START_OBJECT) { + in.nextToken(); + return doMapNext(); + } else { + throw error("map-start"); + } + } + + @Override + public long mapNext() throws IOException { + advance(Symbol.ITEM_END); + return doMapNext(); + } + + private long doMapNext() throws IOException { + if (in.getCurrentToken() == JsonToken.END_OBJECT) { + in.nextToken(); + advance(Symbol.MAP_END); + return 0; + } else { + return 1; + } + } + + @Override + public long skipMap() throws IOException { + advance(Symbol.MAP_START); + if (in.getCurrentToken() == JsonToken.START_OBJECT) { + in.skipChildren(); + in.nextToken(); + advance(Symbol.MAP_END); + } else { + throw error("map-start"); + } + return 0; + } + + @Override + public int readIndex() throws IOException { + advance(Symbol.UNION); + Symbol.Alternative a = (Symbol.Alternative) parser.popSymbol(); + + String label; + final JsonToken currentToken = in.getCurrentToken(); + if (currentToken == JsonToken.VALUE_NULL) { + label = "null"; + } else if (a.size() == 2 && + ("null".equals(a.getLabel(0)) || "null".equals(a.getLabel(1)))) { + label = ("null".equals(a.getLabel(0)) ? a.getLabel(1) : a.getLabel(0)); + } else if (currentToken == JsonToken.START_OBJECT + && in.nextToken() == JsonToken.FIELD_NAME) { + label = in.getText(); + in.nextToken(); + parser.pushSymbol(Symbol.UNION_END); + } else { + throw error("start-union"); + } + + int n = a.findLabel(label); + if (n < 0) { + // in order to allow the type specification for nullable fields to be avoided, we always choose index 1 (second type) + // works if you have only null on index 0, some type on index 1 + // original code threw a new AvroTypeException("Unknown union branch " + label); + n = 1; + } + parser.pushSymbol(a.getSymbol(n)); + return n; + } + + @Override + public Symbol doAction(Symbol input, Symbol top) throws IOException { + if (top instanceof Symbol.FieldAdjustAction) { + Symbol.FieldAdjustAction fa = (Symbol.FieldAdjustAction) top; + String name = fa.fname; + if (currentReorderBuffer != null) { + List node = currentReorderBuffer.savedFields.get(name); + if (node != null) { + currentReorderBuffer.savedFields.remove(name); + currentReorderBuffer.origParser = in; + in = makeParser(node); + return null; + } + } + if (in.getCurrentToken() == JsonToken.FIELD_NAME) { + do { + String fn = in.getText(); + in.nextToken(); + if (name.equals(fn)) { + return null; + } else { + if (currentReorderBuffer == null) { + currentReorderBuffer = new ReorderBuffer(); + } + currentReorderBuffer.savedFields.put(fn, getVaueAsTree(in)); + } + } while (in.getCurrentToken() == JsonToken.FIELD_NAME); + injectDefaultValueIfAvailable(in, fa.fname); + } else { + injectDefaultValueIfAvailable(in, fa.fname); + } + } else if (top == Symbol.FIELD_END) { + if (currentReorderBuffer != null && currentReorderBuffer.origParser != null) { + in = currentReorderBuffer.origParser; + currentReorderBuffer.origParser = null; + } + } else if (top == Symbol.RECORD_START) { + if (in.getCurrentToken() == JsonToken.START_OBJECT) { + in.nextToken(); + reorderBuffers.push(currentReorderBuffer); + currentReorderBuffer = null; + } else { + throw error("record-start"); + } + } else if (top == Symbol.RECORD_END || top == Symbol.UNION_END) { + if (in.getCurrentToken() == JsonToken.END_OBJECT) { + in.nextToken(); + if (top == Symbol.RECORD_END) { + if (currentReorderBuffer != null && !currentReorderBuffer.savedFields.isEmpty()) { + throw error("Unknown fields: " + currentReorderBuffer.savedFields.keySet()); + } + currentReorderBuffer = reorderBuffers.pop(); + } + } else { + throw error(top == Symbol.RECORD_END ? "record-end" : "union-end"); + } + } else { + throw new AvroTypeException("Unknown action symbol " + top); + } + return null; + } + + private static class JsonElement { + public final JsonToken token; + public final String value; + + public JsonElement(JsonToken t, String value) { + this.token = t; + this.value = value; + } + + public JsonElement(JsonToken t) { + this(t, null); + } + } + + private static List getVaueAsTree(JsonParser in) throws IOException { + int level = 0; + List result = new ArrayList(); + do { + JsonToken t = in.getCurrentToken(); + switch (t) { + case START_OBJECT: + case START_ARRAY: + level++; + result.add(new JsonElement(t)); + break; + case END_OBJECT: + case END_ARRAY: + level--; + result.add(new JsonElement(t)); + break; + case FIELD_NAME: + case VALUE_STRING: + case VALUE_NUMBER_INT: + case VALUE_NUMBER_FLOAT: + case VALUE_TRUE: + case VALUE_FALSE: + case VALUE_NULL: + result.add(new JsonElement(t, in.getText())); + break; + } + in.nextToken(); + } while (level != 0); + result.add(new JsonElement(null)); + return result; + } + + private JsonParser makeParser(final List elements) throws IOException { + return new JsonParser() { + int pos = 0; + + @Override + public ObjectCodec getCodec() { + throw new UnsupportedOperationException(); + } + + @Override + public void setCodec(ObjectCodec c) { + throw new UnsupportedOperationException(); + } + + @Override + public void close() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public JsonToken nextToken() throws IOException { + pos++; + return elements.get(pos).token; + } + + @Override + public JsonParser skipChildren() throws IOException { + JsonToken tkn = elements.get(pos).token; + int level = (tkn == JsonToken.START_ARRAY || tkn == JsonToken.END_ARRAY) ? 1 : 0; + while (level > 0) { + switch (elements.get(++pos).token) { + case START_ARRAY: + case START_OBJECT: + level++; + break; + case END_ARRAY: + case END_OBJECT: + level--; + break; + } + } + return this; + } + + @Override + public boolean isClosed() { + throw new UnsupportedOperationException(); + } + + @Override + public String getCurrentName() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public JsonStreamContext getParsingContext() { + throw new UnsupportedOperationException(); + } + + @Override + public JsonLocation getTokenLocation() { + throw new UnsupportedOperationException(); + } + + @Override + public JsonLocation getCurrentLocation() { + throw new UnsupportedOperationException(); + } + + @Override + public String getText() throws IOException { + return elements.get(pos).value; + } + + @Override + public char[] getTextCharacters() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public int getTextLength() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public int getTextOffset() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public Number getNumberValue() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public NumberType getNumberType() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public int getIntValue() throws IOException { + return Integer.parseInt(getText()); + } + + @Override + public long getLongValue() throws IOException { + return Long.parseLong(getText()); + } + + @Override + public BigInteger getBigIntegerValue() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public float getFloatValue() throws IOException { + return Float.parseFloat(getText()); + } + + @Override + public double getDoubleValue() throws IOException { + return Double.parseDouble(getText()); + } + + @Override + public BigDecimal getDecimalValue() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] getBinaryValue(Base64Variant b64variant) + throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public JsonToken getCurrentToken() { + return elements.get(pos).token; + } + }; + } + + private AvroTypeException error(String type) { + return new AvroTypeException("Expected " + type + + ". Got " + in.getCurrentToken()); + } + + private static final JsonElement NULL_JSON_ELEMENT = new JsonElement(null); + + private void injectDefaultValueIfAvailable(final JsonParser in, String fieldName) throws IOException { + Field field = findField(schema, fieldName); + + boolean isNull = field == null; + + JsonNode defVal = isNull ? NullNode.getInstance() : field.defaultValue(); + if (defVal == null) { + throw new AvroTypeException("Expected field name not found: " + fieldName); + } + + List result = new ArrayList<>(2); + JsonParser traverse = defVal.traverse(); + JsonToken nextToken; + while ((nextToken = traverse.nextToken()) != null) { + if (nextToken.isScalarValue()) { + result.add(new JsonElement(nextToken, traverse.getText())); + } else { + result.add(new JsonElement(nextToken)); + } + } + result.add(NULL_JSON_ELEMENT); + if (currentReorderBuffer == null) { + currentReorderBuffer = new ReorderBuffer(); + } + currentReorderBuffer.origParser = in; + this.in = makeParser(result); + } + + private static Field findField(Schema schema, String name) { + if (schema.getField(name) != null) { + return schema.getField(name); + } + + Field foundField = null; + + for (Field field : schema.getFields()) { + Schema fieldSchema = field.schema(); + if (Type.RECORD.equals(fieldSchema.getType())) { + foundField = findField(fieldSchema, name); + } else if (Type.ARRAY.equals(fieldSchema.getType())) { + foundField = findField(fieldSchema.getElementType(), name); + } else if (Type.MAP.equals(fieldSchema.getType())) { + foundField = findField(fieldSchema.getValueType(), name); + } + + if (foundField != null) { + return foundField; + } + } + + return foundField; + } +} +