DGS-8320 Enable defaults for JSON Schema

json-schema-provider/src/main/java/io/confluent/kafka/schemaregistry/json/JsonSchema.java

@@ -404,11 +404,11 @@ public void validate() {
     rawSchema();
   }
 
-  public void validate(Object value) throws JsonProcessingException, ValidationException {
-    validate(rawSchema(), value);
+  public JsonNode validate(Object value) throws JsonProcessingException, ValidationException {
+    return validate(rawSchema(), value);
   }
 
-  public static void validate(Schema schema, Object value)
+  public static JsonNode validate(Schema schema, Object value)
       throws JsonProcessingException, ValidationException {
     Object primitiveValue = NONE_MARKER;
     if (isPrimitive(value)) {
@@ -426,6 +426,7 @@ public static void validate(Schema schema, Object value)
     }
     if (primitiveValue != NONE_MARKER) {
       schema.validate(primitiveValue);
+      return objectMapper.convertValue(primitiveValue, JsonNode.class);
     } else {
       Object jsonObject;
       if (value instanceof ArrayNode) {
@@ -438,6 +439,7 @@ public static void validate(Schema schema, Object value)
         jsonObject = objectMapper.convertValue(value, JSONObject.class);
       }
       schema.validate(jsonObject);
+      return objectMapper.convertValue(jsonObject, JsonNode.class);
     }
   }
 

json-schema-provider/src/main/java/io/confluent/kafka/schemaregistry/json/JsonSchemaUtils.java

@@ -258,7 +258,7 @@ public static Object toObject(JsonNode value, JsonSchema schema) throws IOExcept
   public static Object toObject(JsonNode value, JsonSchema schema, boolean validate)
       throws IOException {
     if (validate) {
-      schema.validate(value);
+      value = schema.validate(value);
     }
     return envelope(schema, value);
   }

json-schema-provider/src/test/java/io/confluent/kafka/schemaregistry/json/JsonSchemaTest.java

@@ -67,6 +67,18 @@ public class JsonSchemaTest {
 
   private static final JsonSchema recordSchema = new JsonSchema(recordSchemaString);
 
+  private static final String recordWithDefaultsSchemaString = "{\"properties\": {\n"
+      + "     \"null\": {\"type\": \"null\", \"default\": null},\n"
+      + "     \"boolean\": {\"type\": \"boolean\", \"default\": true},\n"
+      + "     \"number\": {\"type\": \"number\", \"default\": 123},\n"
+      + "     \"string\": {\"type\": \"string\", \"default\": \"abc\"}\n"
+      + "  },\n"
+      + "  \"additionalProperties\": false\n"
+      + "}";
+
+  private static final JsonSchema recordWithDefaultsSchema =
+      new JsonSchema(recordWithDefaultsSchemaString);
+
   private static final String arraySchemaString = "{\"type\": \"array\", \"items\": { \"type\": "
       + "\"string\" } }";
 
@@ -136,6 +148,18 @@ public void testRecordToJsonSchema() throws Exception {
     assertEquals("string", result.get("string").textValue());
   }
 
+  @Test
+  public void testRecordWithDefaultsToJsonSchema() throws Exception {
+    String json = "{}";
+
+    JsonNode envelope = (JsonNode) JsonSchemaUtils.toObject(json, recordWithDefaultsSchema);
+    JsonNode result = (JsonNode) JsonSchemaUtils.getValue(envelope);
+    assertEquals(true, result.get("null").isNull());
+    assertEquals(true, result.get("boolean").booleanValue());
+    assertEquals(123, result.get("number").intValue());
+    assertEquals("abc", result.get("string").textValue());
+  }
+
   @Test(expected = ValidationException.class)
   public void testInvalidRecordToJsonSchema() throws Exception {
     String json = "{\n"

json-schema-serializer/src/main/java/io/confluent/kafka/serializers/json/AbstractKafkaJsonSchemaDeserializer.java

@@ -175,7 +175,7 @@ protected Object deserialize(
           if (jsonNode == null) {
             jsonNode = objectMapper.readValue(buffer.array(), start, length, JsonNode.class);
           }
-          schema.validate(jsonNode);
+          jsonNode = schema.validate(jsonNode);
         } catch (JsonProcessingException | ValidationException e) {
           throw new SerializationException("JSON "
               + jsonNode

json-schema-serializer/src/main/java/io/confluent/kafka/serializers/json/AbstractKafkaJsonSchemaSerializer.java

@@ -156,7 +156,7 @@ protected byte[] serializeImpl(
       }
       object = (T) executeRules(subject, topic, headers, RuleMode.WRITE, null, schema, object);
       if (validate) {
-        validateJson(object, schema);
+        object = validateJson(object, schema);
       }
       ByteArrayOutputStream out = new ByteArrayOutputStream();
       out.write(MAGIC_BYTE);
@@ -176,12 +176,14 @@ protected byte[] serializeImpl(
     }
   }
 
-  protected void validateJson(T object,
-                              JsonSchema schema)
+  @SuppressWarnings("unchecked")
+  protected T validateJson(T object,
+                           JsonSchema schema)
       throws SerializationException {
     try {
       JsonNode jsonNode = objectMapper.convertValue(object, JsonNode.class);
-      schema.validate(jsonNode);
+      jsonNode = schema.validate(jsonNode);
+      return (T) objectMapper.convertValue(jsonNode, object.getClass());
     } catch (JsonProcessingException e) {
       throw new SerializationException("JSON "
           + object

json-schema-serializer/src/test/java/io/confluent/kafka/serializers/json/KafkaJsonSchemaSerializerTest.java

@@ -16,6 +16,8 @@
 package io.confluent.kafka.serializers.json;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject;
@@ -47,6 +49,18 @@
 
 public class KafkaJsonSchemaSerializerTest {
 
+  private static final String recordWithDefaultsSchemaString = "{\"properties\": {\n"
+      + "     \"null\": {\"type\": \"null\", \"default\": null},\n"
+      + "     \"boolean\": {\"type\": \"boolean\", \"default\": true},\n"
+      + "     \"number\": {\"type\": \"number\", \"default\": 123},\n"
+      + "     \"string\": {\"type\": \"string\", \"default\": \"abc\"}\n"
+      + "  },\n"
+      + "  \"additionalProperties\": false\n"
+      + "}";
+
+  private static final JsonSchema recordWithDefaultsSchema =
+      new JsonSchema(recordWithDefaultsSchemaString);
+
   private final Properties config;
   private final SchemaRegistryClient schemaRegistry;
   private KafkaJsonSchemaSerializer<Object> serializer;
@@ -216,6 +230,25 @@ public void serializeUserRef() throws Exception {
     assertEquals(customer, deserialized);
   }
 
+  @Test
+  public void serializeRecordWithDefaults() throws Exception {
+    schemaRegistry.register(topic + "-value", recordWithDefaultsSchema);
+
+    String json = "{}";
+    JsonNode record = new ObjectMapper().readTree(json);
+    byte[] bytes = latestSerializer.serialize(topic, record);
+
+    String expectedJson = "{\n"
+        + "    \"null\": null,\n"
+        + "    \"boolean\": true,\n"
+        + "    \"number\": 123,\n"
+        + "    \"string\": \"abc\"\n"
+        + "}";
+    JsonNode expectedRecord = new ObjectMapper().readTree(expectedJson);
+    Object deserialized = getDeserializer(null).deserialize(topic, bytes);
+    assertEquals(expectedRecord, deserialized);
+  }
+
   // Generate javaType property
   @JsonSchemaInject(strings = {@JsonSchemaString(path="javaType",
       value="io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializerTest$Customer")})