You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ce...@apache.org on 2023/06/02 15:20:53 UTC
[kafka] branch trunk updated: KAFKA-15012: Allow leading zeros in numeric fields while deserializing JSON messages using the JsonConverter (#13800)
This is an automated email from the ASF dual-hosted git repository.
cegerton pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 02fb4b882b8 KAFKA-15012: Allow leading zeros in numeric fields while deserializing JSON messages using the JsonConverter (#13800)
02fb4b882b8 is described below
commit 02fb4b882b86bc835cf4b0bbdaee099d6056799b
Author: Yash Mayya <ya...@gmail.com>
AuthorDate: Fri Jun 2 16:20:37 2023 +0100
KAFKA-15012: Allow leading zeros in numeric fields while deserializing JSON messages using the JsonConverter (#13800)
Reviewers: Sagar Rao <sa...@gmail.com>, Chris Egerton <ch...@aiven.io>
---
.../main/java/org/apache/kafka/connect/json/JsonDeserializer.java | 7 +++++--
.../java/org/apache/kafka/connect/json/JsonConverterTest.java | 8 ++++++++
2 files changed, 13 insertions(+), 2 deletions(-)
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 970091d47b3..90f81f15237 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
@@ -16,15 +16,17 @@
*/
package org.apache.kafka.connect.json;
+import com.fasterxml.jackson.core.json.JsonReadFeature;
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;
import org.apache.kafka.common.serialization.Deserializer;
+import java.util.Collections;
+import java.util.Set;
+
/**
* JSON deserializer for Jackson's {@link JsonNode} tree model. Using the tree model allows it to work with arbitrarily
* structured data without having associated Java classes. This deserializer also supports Connect schemas.
@@ -50,6 +52,7 @@ public class JsonDeserializer implements Deserializer<JsonNode> {
final Set<DeserializationFeature> deserializationFeatures,
final JsonNodeFactory jsonNodeFactory
) {
+ objectMapper.enable(JsonReadFeature.ALLOW_LEADING_ZEROS_FOR_NUMBERS.mappedFeature());
deserializationFeatures.forEach(objectMapper::enable);
objectMapper.setNodeFactory(jsonNodeFactory);
}
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 d3d8999cc53..f9bc2c67766 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
@@ -115,6 +115,14 @@ public class JsonConverterTest {
assertEquals(new SchemaAndValue(Schema.INT64_SCHEMA, 4398046511104L), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"int64\" }, \"payload\": 4398046511104 }".getBytes()));
}
+ @Test
+ public void numberWithLeadingZerosToConnect() {
+ assertEquals(new SchemaAndValue(Schema.INT8_SCHEMA, (byte) 12), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"int8\" }, \"payload\": 0012 }".getBytes()));
+ assertEquals(new SchemaAndValue(Schema.INT16_SCHEMA, (short) 123), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"int16\" }, \"payload\": 000123 }".getBytes()));
+ assertEquals(new SchemaAndValue(Schema.INT32_SCHEMA, 12345), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"int32\" }, \"payload\": 000012345 }".getBytes()));
+ assertEquals(new SchemaAndValue(Schema.INT64_SCHEMA, 123456789L), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"int64\" }, \"payload\": 00000123456789 }".getBytes()));
+ }
+
@Test
public void floatToConnect() {
assertEquals(new SchemaAndValue(Schema.FLOAT32_SCHEMA, 12.34f), converter.toConnectData(TOPIC, "{ \"schema\": { \"type\": \"float\" }, \"payload\": 12.34 }".getBytes()));