kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rha...@apache.org
Subject [kafka] branch 2.4 updated: KAFKA-9667: Connect JSON serde strip trailing zeros (#8230)
Date Thu, 07 May 2020 22:41:44 GMT
This is an automated email from the ASF dual-hosted git repository.

rhauch pushed a commit to branch 2.4
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.4 by this push:
     new 2a80b72  KAFKA-9667: Connect JSON serde strip trailing zeros (#8230)
2a80b72 is described below

commit 2a80b7299366640b1ae266cc0d1b94d4a4726f28
Author: Andy Coates <8012398+big-andy-coates@users.noreply.github.com>
AuthorDate: Thu May 7 22:21:08 2020 +0100

    KAFKA-9667: Connect JSON serde strip trailing zeros (#8230)
    
    This change turns on exact decimal processing in JSON Converter for deserializing decimals,
meaning trailing zeros are maintained. Serialization was already using the decimal scale to
output the right value, so this change means a value of `1.2300` can now be serialized to
JSON and deserialized back to Connect without any loss of information.
    
    Author: Andy Coates <big-andy-coates@users.noreply.github.com>
    Reviewers: Randall Hauch <rhauch@gmail.com>, Almog Gavra <almog@confluent.io>
---
 .../apache/kafka/connect/json/JsonConverter.java   | 81 ++++++++++++----------
 .../kafka/connect/json/JsonDeserializer.java       | 12 +++-
 .../apache/kafka/connect/json/JsonSerializer.java  | 20 ++++++
 .../kafka/connect/json/JsonConverterTest.java      | 31 ++++++++-
 4 files changed, 103 insertions(+), 41 deletions(-)

diff --git a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
index 3ef6aae..7af202b 100644
--- a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
+++ b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
@@ -21,8 +21,6 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
-import java.util.HashSet;
-import java.util.Set;
 import org.apache.kafka.common.cache.Cache;
 import org.apache.kafka.common.cache.LRUCache;
 import org.apache.kafka.common.cache.SynchronizedCache;
@@ -54,6 +52,8 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 
+import static org.apache.kafka.common.utils.Utils.mkSet;
+
 /**
  * Implementation of Converter that uses JSON to store schemas and objects. By default this
converter will serialize Connect keys, values,
  * and headers with schemas, although this can be disabled with {@link JsonConverterConfig#SCHEMAS_ENABLE_CONFIG
schemas.enable}
@@ -191,6 +191,9 @@ public class JsonConverter implements Converter, HeaderConverter {
     // Convert values in Kafka Connect form into/from their logical types. These logical
converters are discovered by logical type
     // names specified in the field
     private static final HashMap<String, LogicalTypeConverter> LOGICAL_CONVERTERS =
new HashMap<>();
+
+    private static final JsonNodeFactory JSON_NODE_FACTORY = JsonNodeFactory.withExactBigDecimals(true);
+
     static {
         LOGICAL_CONVERTERS.put(Decimal.LOGICAL_NAME, new LogicalTypeConverter() {
             @Override
@@ -201,9 +204,9 @@ public class JsonConverter implements Converter, HeaderConverter {
                 final BigDecimal decimal = (BigDecimal) value;
                 switch (config.decimalFormat()) {
                     case NUMERIC:
-                        return JsonNodeFactory.instance.numberNode(decimal);
+                        return JSON_NODE_FACTORY.numberNode(decimal);
                     case BASE64:
-                        return JsonNodeFactory.instance.binaryNode(Decimal.fromLogical(schema,
decimal));
+                        return JSON_NODE_FACTORY.binaryNode(Decimal.fromLogical(schema, decimal));
                     default:
                         throw new DataException("Unexpected " + JsonConverterConfig.DECIMAL_FORMAT_CONFIG
+ ": " + config.decimalFormat());
                 }
@@ -229,7 +232,7 @@ public class JsonConverter implements Converter, HeaderConverter {
             public JsonNode toJson(final Schema schema, final Object value, final JsonConverterConfig
config) {
                 if (!(value instanceof java.util.Date))
                     throw new DataException("Invalid type for Date, expected Date but was
" + value.getClass());
-                return JsonNodeFactory.instance.numberNode(Date.fromLogical(schema, (java.util.Date)
value));
+                return JSON_NODE_FACTORY.numberNode(Date.fromLogical(schema, (java.util.Date)
value));
             }
 
             @Override
@@ -245,7 +248,7 @@ public class JsonConverter implements Converter, HeaderConverter {
             public JsonNode toJson(final Schema schema, final Object value, final JsonConverterConfig
config) {
                 if (!(value instanceof java.util.Date))
                     throw new DataException("Invalid type for Time, expected Date but was
" + value.getClass());
-                return JsonNodeFactory.instance.numberNode(Time.fromLogical(schema, (java.util.Date)
value));
+                return JSON_NODE_FACTORY.numberNode(Time.fromLogical(schema, (java.util.Date)
value));
             }
 
             @Override
@@ -261,7 +264,7 @@ public class JsonConverter implements Converter, HeaderConverter {
             public JsonNode toJson(final Schema schema, final Object value, final JsonConverterConfig
config) {
                 if (!(value instanceof java.util.Date))
                     throw new DataException("Invalid type for Timestamp, expected Date but
was " + value.getClass());
-                return JsonNodeFactory.instance.numberNode(Timestamp.fromLogical(schema,
(java.util.Date) value));
+                return JSON_NODE_FACTORY.numberNode(Timestamp.fromLogical(schema, (java.util.Date)
value));
             }
 
             @Override
@@ -277,15 +280,23 @@ public class JsonConverter implements Converter, HeaderConverter {
     private Cache<Schema, ObjectNode> fromConnectSchemaCache;
     private Cache<JsonNode, Schema> toConnectSchemaCache;
 
-    private final JsonSerializer serializer = new JsonSerializer();
+    private final JsonSerializer serializer;
     private final JsonDeserializer deserializer;
 
     public JsonConverter() {
-        // this ensures that the JsonDeserializer maintains full precision on
-        // floating point numbers that cannot fit into float64
-        final Set<DeserializationFeature> deserializationFeatures = new HashSet<>();
-        deserializationFeatures.add(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
-        deserializer = new JsonDeserializer(deserializationFeatures);
+        serializer = new JsonSerializer(
+            mkSet(),
+            JSON_NODE_FACTORY
+        );
+
+        deserializer = new JsonDeserializer(
+            mkSet(
+                // this ensures that the JsonDeserializer maintains full precision on
+                // floating point numbers that cannot fit into float64
+                DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS
+            ),
+            JSON_NODE_FACTORY
+        );
     }
 
     @Override
@@ -362,7 +373,7 @@ public class JsonConverter implements Converter, HeaderConverter {
         // The deserialized data should either be an envelope object containing the schema
and the payload or the schema
         // was stripped during serialization and we need to fill in an all-encompassing schema.
         if (!config.schemasEnabled()) {
-            ObjectNode envelope = JsonNodeFactory.instance.objectNode();
+            ObjectNode envelope = JSON_NODE_FACTORY.objectNode();
             envelope.set(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME, null);
             envelope.set(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME, jsonValue);
             jsonValue = envelope;
@@ -413,17 +424,17 @@ public class JsonConverter implements Converter, HeaderConverter {
                 jsonSchema = JsonSchema.STRING_SCHEMA.deepCopy();
                 break;
             case ARRAY:
-                jsonSchema = JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME,
JsonSchema.ARRAY_TYPE_NAME);
+                jsonSchema = JSON_NODE_FACTORY.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME,
JsonSchema.ARRAY_TYPE_NAME);
                 jsonSchema.set(JsonSchema.ARRAY_ITEMS_FIELD_NAME, asJsonSchema(schema.valueSchema()));
                 break;
             case MAP:
-                jsonSchema = JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME,
JsonSchema.MAP_TYPE_NAME);
+                jsonSchema = JSON_NODE_FACTORY.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME,
JsonSchema.MAP_TYPE_NAME);
                 jsonSchema.set(JsonSchema.MAP_KEY_FIELD_NAME, asJsonSchema(schema.keySchema()));
                 jsonSchema.set(JsonSchema.MAP_VALUE_FIELD_NAME, asJsonSchema(schema.valueSchema()));
                 break;
             case STRUCT:
-                jsonSchema = JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME,
JsonSchema.STRUCT_TYPE_NAME);
-                ArrayNode fields = JsonNodeFactory.instance.arrayNode();
+                jsonSchema = JSON_NODE_FACTORY.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME,
JsonSchema.STRUCT_TYPE_NAME);
+                ArrayNode fields = JSON_NODE_FACTORY.arrayNode();
                 for (Field field : schema.fields()) {
                     ObjectNode fieldJsonSchema = asJsonSchema(field.schema()).deepCopy();
                     fieldJsonSchema.put(JsonSchema.STRUCT_FIELD_NAME_FIELD_NAME, field.name());
@@ -443,7 +454,7 @@ public class JsonConverter implements Converter, HeaderConverter {
         if (schema.doc() != null)
             jsonSchema.put(JsonSchema.SCHEMA_DOC_FIELD_NAME, schema.doc());
         if (schema.parameters() != null) {
-            ObjectNode jsonSchemaParams = JsonNodeFactory.instance.objectNode();
+            ObjectNode jsonSchemaParams = JSON_NODE_FACTORY.objectNode();
             for (Map.Entry<String, String> prop : schema.parameters().entrySet())
                 jsonSchemaParams.put(prop.getKey(), prop.getValue());
             jsonSchema.set(JsonSchema.SCHEMA_PARAMETERS_FIELD_NAME, jsonSchemaParams);
@@ -596,7 +607,7 @@ public class JsonConverter implements Converter, HeaderConverter {
             if (schema.defaultValue() != null)
                 return convertToJson(schema, schema.defaultValue());
             if (schema.isOptional())
-                return JsonNodeFactory.instance.nullNode();
+                return JSON_NODE_FACTORY.nullNode();
             throw new DataException("Conversion error: null value for field that is required
and has no default value");
         }
 
@@ -617,32 +628,32 @@ public class JsonConverter implements Converter, HeaderConverter {
             }
             switch (schemaType) {
                 case INT8:
-                    return JsonNodeFactory.instance.numberNode((Byte) value);
+                    return JSON_NODE_FACTORY.numberNode((Byte) value);
                 case INT16:
-                    return JsonNodeFactory.instance.numberNode((Short) value);
+                    return JSON_NODE_FACTORY.numberNode((Short) value);
                 case INT32:
-                    return JsonNodeFactory.instance.numberNode((Integer) value);
+                    return JSON_NODE_FACTORY.numberNode((Integer) value);
                 case INT64:
-                    return JsonNodeFactory.instance.numberNode((Long) value);
+                    return JSON_NODE_FACTORY.numberNode((Long) value);
                 case FLOAT32:
-                    return JsonNodeFactory.instance.numberNode((Float) value);
+                    return JSON_NODE_FACTORY.numberNode((Float) value);
                 case FLOAT64:
-                    return JsonNodeFactory.instance.numberNode((Double) value);
+                    return JSON_NODE_FACTORY.numberNode((Double) value);
                 case BOOLEAN:
-                    return JsonNodeFactory.instance.booleanNode((Boolean) value);
+                    return JSON_NODE_FACTORY.booleanNode((Boolean) value);
                 case STRING:
                     CharSequence charSeq = (CharSequence) value;
-                    return JsonNodeFactory.instance.textNode(charSeq.toString());
+                    return JSON_NODE_FACTORY.textNode(charSeq.toString());
                 case BYTES:
                     if (value instanceof byte[])
-                        return JsonNodeFactory.instance.binaryNode((byte[]) value);
+                        return JSON_NODE_FACTORY.binaryNode((byte[]) value);
                     else if (value instanceof ByteBuffer)
-                        return JsonNodeFactory.instance.binaryNode(((ByteBuffer) value).array());
+                        return JSON_NODE_FACTORY.binaryNode(((ByteBuffer) value).array());
                     else
                         throw new DataException("Invalid type for bytes type: " + value.getClass());
                 case ARRAY: {
                     Collection collection = (Collection) value;
-                    ArrayNode list = JsonNodeFactory.instance.arrayNode();
+                    ArrayNode list = JSON_NODE_FACTORY.arrayNode();
                     for (Object elem : collection) {
                         Schema valueSchema = schema == null ? null : schema.valueSchema();
                         JsonNode fieldValue = convertToJson(valueSchema, elem);
@@ -668,9 +679,9 @@ public class JsonConverter implements Converter, HeaderConverter {
                     ObjectNode obj = null;
                     ArrayNode list = null;
                     if (objectMode)
-                        obj = JsonNodeFactory.instance.objectNode();
+                        obj = JSON_NODE_FACTORY.objectNode();
                     else
-                        list = JsonNodeFactory.instance.arrayNode();
+                        list = JSON_NODE_FACTORY.arrayNode();
                     for (Map.Entry<?, ?> entry : map.entrySet()) {
                         Schema keySchema = schema == null ? null : schema.keySchema();
                         Schema valueSchema = schema == null ? null : schema.valueSchema();
@@ -680,7 +691,7 @@ public class JsonConverter implements Converter, HeaderConverter {
                         if (objectMode)
                             obj.set(mapKey.asText(), mapValue);
                         else
-                            list.add(JsonNodeFactory.instance.arrayNode().add(mapKey).add(mapValue));
+                            list.add(JSON_NODE_FACTORY.arrayNode().add(mapKey).add(mapValue));
                     }
                     return objectMode ? obj : list;
                 }
@@ -688,7 +699,7 @@ public class JsonConverter implements Converter, HeaderConverter {
                     Struct struct = (Struct) value;
                     if (!struct.schema().equals(schema))
                         throw new DataException("Mismatching schema.");
-                    ObjectNode obj = JsonNodeFactory.instance.objectNode();
+                    ObjectNode obj = JSON_NODE_FACTORY.objectNode();
                     for (Field field : schema.fields()) {
                         obj.set(field.name(), convertToJson(field.schema(), struct.get(field)));
                     }
diff --git a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonDeserializer.java
b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonDeserializer.java
index a656e53..2e6e821 100644
--- a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonDeserializer.java
+++ b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonDeserializer.java
@@ -19,6 +19,7 @@ package org.apache.kafka.connect.json;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import java.util.Collections;
 import java.util.Set;
 import org.apache.kafka.common.errors.SerializationException;
@@ -29,13 +30,13 @@ import org.apache.kafka.common.serialization.Deserializer;
  * structured data without having associated Java classes. This deserializer also supports
Connect schemas.
  */
 public class JsonDeserializer implements Deserializer<JsonNode> {
-    private ObjectMapper objectMapper = new ObjectMapper();
+    private final ObjectMapper objectMapper = new ObjectMapper();
 
     /**
      * Default constructor needed by Kafka
      */
     public JsonDeserializer() {
-        this(Collections.emptySet());
+        this(Collections.emptySet(), JsonNodeFactory.withExactBigDecimals(true));
     }
 
     /**
@@ -43,9 +44,14 @@ public class JsonDeserializer implements Deserializer<JsonNode> {
      * for the deserializer
      *
      * @param deserializationFeatures the specified deserialization features
+     * @param jsonNodeFactory the json node factory to use.
      */
-    JsonDeserializer(final Set<DeserializationFeature> deserializationFeatures) {
+    JsonDeserializer(
+        final Set<DeserializationFeature> deserializationFeatures,
+        final JsonNodeFactory jsonNodeFactory
+    ) {
         deserializationFeatures.forEach(objectMapper::enable);
+        objectMapper.setNodeFactory(jsonNodeFactory);
     }
 
     @Override
diff --git a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonSerializer.java
b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonSerializer.java
index 94ec0a8..0f2b62b 100644
--- a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonSerializer.java
+++ b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonSerializer.java
@@ -18,9 +18,14 @@ package org.apache.kafka.connect.json;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.serialization.Serializer;
 
+import java.util.Collections;
+import java.util.Set;
+
 /**
  * Serialize Jackson JsonNode tree model objects to UTF-8 JSON. Using the tree model allows
handling arbitrarily
  * structured data without corresponding Java classes. This serializer also supports Connect
schemas.
@@ -32,7 +37,22 @@ public class JsonSerializer implements Serializer<JsonNode> {
      * Default constructor needed by Kafka
      */
     public JsonSerializer() {
+        this(Collections.emptySet(), JsonNodeFactory.withExactBigDecimals(true));
+    }
 
+    /**
+     * A constructor that additionally specifies some {@link SerializationFeature}
+     * for the serializer
+     *
+     * @param serializationFeatures the specified serialization features
+     * @param jsonNodeFactory the json node factory to use.
+     */
+    JsonSerializer(
+        final Set<SerializationFeature> serializationFeatures,
+        final JsonNodeFactory jsonNodeFactory
+    ) {
+        serializationFeatures.forEach(objectMapper::enable);
+        objectMapper.setNodeFactory(jsonNodeFactory);
     }
 
     @Override
diff --git a/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java
b/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java
index 2a56950..2e189e2 100644
--- a/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java
+++ b/connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java
@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.connect.json;
 
+import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
@@ -65,8 +66,11 @@ import static org.junit.Assert.fail;
 public class JsonConverterTest {
     private static final String TOPIC = "topic";
 
-    ObjectMapper objectMapper = new ObjectMapper();
-    JsonConverter converter = new JsonConverter();
+    private final ObjectMapper objectMapper = new ObjectMapper()
+        .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS)
+        .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true));
+
+    private final JsonConverter converter = new JsonConverter();
 
     @Before
     public void setUp() {
@@ -273,6 +277,16 @@ public class JsonConverterTest {
     }
 
     @Test
+    public void numericDecimalWithTrailingZerosToConnect() {
+        BigDecimal reference = new BigDecimal(new BigInteger("15600"), 4);
+        Schema schema = Decimal.schema(4);
+        String msg = "{ \"schema\": { \"type\": \"bytes\", \"name\": \"org.apache.kafka.connect.data.Decimal\",
\"version\": 1, \"parameters\": { \"scale\": \"4\" } }, \"payload\": 1.5600 }";
+        SchemaAndValue schemaAndValue = converter.toConnectData(TOPIC, msg.getBytes());
+        assertEquals(schema, schemaAndValue.schema());
+        assertEquals(reference, schemaAndValue.value());
+    }
+
+    @Test
     public void highPrecisionNumericDecimalToConnect() {
         // this number is too big to be kept in a float64!
         BigDecimal reference = new BigDecimal("1.23456789123456789");
@@ -634,7 +648,18 @@ public class JsonConverterTest {
     }
 
     @Test
-    public void decimalToJsonWithoutSchema() throws IOException {
+    public void decimalWithTrailingZerosToNumericJson() {
+        converter.configure(Collections.singletonMap(JsonConverterConfig.DECIMAL_FORMAT_CONFIG,
DecimalFormat.NUMERIC.name()), false);
+        JsonNode converted = parse(converter.fromConnectData(TOPIC, Decimal.schema(4), new
BigDecimal(new BigInteger("15600"), 4)));
+        validateEnvelope(converted);
+        assertEquals(parse("{ \"type\": \"bytes\", \"optional\": false, \"name\": \"org.apache.kafka.connect.data.Decimal\",
\"version\": 1, \"parameters\": { \"scale\": \"4\" } }"),
+            converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
+        assertTrue("expected node to be numeric", converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).isNumber());
+        assertEquals(new BigDecimal("1.5600"), converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).decimalValue());
+    }
+
+    @Test
+    public void decimalToJsonWithoutSchema() {
         assertThrows(
             "expected data exception when serializing BigDecimal without schema",
             DataException.class,


Mime
View raw message