fix(filters): fix JSONFilter not supporting null field (#71)

This commit improves support for NULL field with no explicit schema.

Resolves: #71

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/data/TypedValue.java

@@ -358,7 +358,8 @@ public boolean isEmpty() {
     }
 
     public TypedValue as(final Type type) {
-        return TypedValue.of(type.convert(value), type);
+        Object converted = isNull() ? null : type.convert(value);
+        return TypedValue.of(converted, type);
     }
 
     /**

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/internal/ConnectSchemaMapper.java

@@ -96,11 +96,13 @@ public Schema map(final StructSchema schema) {
 
         for(final TypedField field : schema) {
             final io.streamthoughts.kafka.connect.filepulse.data.Schema fieldSchema = field.schema();
-            final String fieldName = field.name();
-            mayUpdateSchemaName(fieldSchema, fieldName);
-            sb.field(fieldName, fieldSchema.map(this)).optional();
+            // Ignore schema NULL because cannot determine the expected type.
+            if (fieldSchema.type() != Type.NULL) {
+                final String fieldName = field.name();
+                mayUpdateSchemaName(fieldSchema, fieldName);
+                sb.field(fieldName, fieldSchema.map(this)).optional();
+            }
         }
-
         return sb.build();
     }
 

connect-file-pulse-api/src/test/java/io/streamthoughts/kafka/connect/filepulse/source/internal/ConnectSchemaMapperTest.java

@@ -18,6 +18,7 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.source.internal;
 
+import io.streamthoughts.kafka.connect.filepulse.data.Schema;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.data.Struct;
@@ -124,4 +125,18 @@ public void shouldMapGivenTypeStructWithMapWithStructValue() {
         Assert.assertEquals("value", field1.get("field2").getString("field3"));
     }
 
+    @Test
+    public void shouldMapGivenTypeStructWithNullValue() {
+        TypedStruct struct = TypedStruct.create()
+                .put("field1", "value1")
+                .put("field2", Schema.none(), null);
+
+        SchemaAndValue schemaAndValue = struct.schema().map(ConnectSchemaMapper.INSTANCE, struct);
+        Assert.assertNotNull(schemaAndValue);
+
+        Struct connectStruct = (Struct)schemaAndValue.value();
+        Assert.assertNotNull(connectStruct.schema().field("field1"));
+        Assert.assertNull(connectStruct.schema().field("field2"));
+    }
+
 }

connect-file-pulse-expression/src/main/java/io/streamthoughts/kafka/connect/filepulse/expression/function/impl/Converts.java

@@ -57,7 +57,6 @@ public SimpleArguments prepare(final TypedValue[] args) {
      */
     @Override
     public TypedValue apply(final TypedValue field, final SimpleArguments args) {
-        Type type = args.valueOf(TYPE);
-        return TypedValue.of(type.convert(field.value()), type);
+        return field.as(args.valueOf(TYPE));
     }
 }

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/json/DefaultJSONStructConverter.java

@@ -21,6 +21,7 @@
 import com.jsoniter.JsonIterator;
 import com.jsoniter.ValueType;
 import com.jsoniter.any.Any;
+import io.streamthoughts.kafka.connect.filepulse.data.Schema;
 import io.streamthoughts.kafka.connect.filepulse.data.Type;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
@@ -45,6 +46,7 @@ public DefaultJSONStructConverter() {
         ACCESSORS.put(ValueType.OBJECT, new ObjectJsonFieldAccessor());
         ACCESSORS.put(ValueType.BOOLEAN, new BooleanJsonFieldAccessor());
         ACCESSORS.put(ValueType.NUMBER, new NumberJsonFieldAccessor());
+        ACCESSORS.put(ValueType.NULL, new NullJsonFieldAccessor());
     }
 
     private static JsonFieldAccessor<?> getAccessorForType(final ValueType type) {
@@ -83,6 +85,14 @@ private interface JsonFieldAccessor<T> {
         TypedValue read(final JsonIterator it) throws IOException;
     }
 
+    private static class NullJsonFieldAccessor implements JsonFieldAccessor<Object> {
+
+        @Override
+        public TypedValue read(JsonIterator it) {
+            return TypedValue.of(null, Schema.none());
+        }
+    }
+
     private static class NumberJsonFieldAccessor implements JsonFieldAccessor<Object> {
 
         @Override

connect-file-pulse-filters/src/test/java/io/streamthoughts/kafka/connect/filepulse/json/DefaultJSONStructConverterTest.java

@@ -127,4 +127,18 @@ public void shouldConvertGivenFieldsWithNumberType() throws Exception {
         assertEquals(Float.MAX_VALUE, struct.getDouble("field-float").floatValue(), 0.0);
     }
 
+    @Test
+    public void shouldConvertGivenFieldWithNullValue() throws Exception {
+        TypedValue value = converter.readJson("{\"field-one\" : \"one\", \"field-two\":null}");
+        Assert.assertNotNull(value);
+        Assert.assertEquals(Type.STRUCT, value.type());
+        TypedStruct struct = value.getStruct();
+
+        StructSchema schema = struct.schema();
+        assertEquals(2, schema.fields().size());
+        assertNotNull(schema.field("field-one"));
+        assertEquals(Type.STRING, schema.field("field-one").schema().type());
+        assertNotNull(schema.field("field-two"));
+        assertEquals(Type.NULL, schema.field("field-two").schema().type());
+    }
 }