fix(filter): fix JSONFilter throwns NPE when parsing empty Array (#109)

Resolves: GH-109

Author: fhussonnois

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

@@ -58,11 +58,22 @@ public <T> T map(final SchemaMapper<T> mapper) {
         return mapper.map(this);
     }
 
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public <T> T map(final SchemaMapperWithValue<T> mapper, final Object object) {
         return mapper.map(this, (Collection)object);
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean isResolvable() {
+        return valueSchema.isResolvable();
+    }
+
     /**
      * {@inheritDoc}
      */

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

@@ -51,4 +51,12 @@ public Schema valueSchema() {
         }
         return valueSchema;
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean isResolvable() {
+        return valueSchema != null || !list.isEmpty();
+    }
 }

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

@@ -35,6 +35,9 @@ public class LazyMapSchema extends MapSchema implements Schema {
 
     }
 
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public Schema valueSchema() {
         if (valueSchema == null) {
@@ -46,4 +49,12 @@ public Schema valueSchema() {
         }
         return valueSchema;
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean isResolvable() {
+        return valueSchema!= null || !map.isEmpty();
+    }
 }

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

@@ -65,11 +65,22 @@ public <T> T map(final SchemaMapper<T> mapper) {
         return mapper.map(this);
     }
 
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public <T> T map(final SchemaMapperWithValue<T> mapper, final Object object) {
         return mapper.map(this, (Map<String, ?>)object);
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean isResolvable() {
+        return valueSchema.isResolvable();
+    }
+
     /**
      * {@inheritDoc}
      */

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

@@ -152,6 +152,17 @@ static ArraySchema array(final Collection<?> value, final Schema valueSchema) {
      */
     Type type() ;
 
+    /**
+     * Checks whether this schemas is resolvable.
+     *
+     * @see LazyArraySchema
+     * @see LazyMapSchema
+     * @return  {@code true}.
+     */
+    default boolean isResolvable() {
+        return true;
+    }
+
     /**
      * Maps this schema into a new type T.
      *

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

@@ -34,6 +34,8 @@
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.connect.data.Struct;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -44,6 +46,8 @@
 
 public class ConnectSchemaMapper implements SchemaMapper<Schema>, SchemaMapperWithValue<SchemaAndValue> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(ConnectSchemaMapper.class);
+
     private static final Object DEFAULT_NULL_VALUE = null;
 
     public static final ConnectSchemaMapper INSTANCE = new ConnectSchemaMapper();
@@ -96,12 +100,14 @@ public Schema map(final StructSchema schema) {
         }
 
         for(final TypedField field : schema) {
+            final String fieldName = field.name();
             final io.streamthoughts.kafka.connect.filepulse.data.Schema fieldSchema = field.schema();
             // Ignore schema NULL because cannot determine the expected type.
-            if (fieldSchema.type() != Type.NULL) {
-                final String fieldName = field.name();
+            if (fieldSchema.type() != Type.NULL && fieldSchema.isResolvable()) {
                 mayUpdateSchemaName(fieldSchema, fieldName);
                 sb.field(fieldName, fieldSchema.map(this)).optional();
+            } else {
+                LOG.debug("Ignore field '{}', schema type is either NULL or cannot be resolved.", fieldName);
             }
         }
         return sb.build();

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

@@ -32,7 +32,7 @@
 public class ConnectSchemaMapperTest {
 
     @Test
-    public void shouldMapGivenSimpleTypedStruct() {
+    public void should_map_given_simple_typed_struct() {
 
         TypedStruct struct = TypedStruct.create()
                 .put("field1", "value1")
@@ -47,7 +47,7 @@ public void shouldMapGivenSimpleTypedStruct() {
     }
 
     @Test
-    public void shouldMapGivenNestedTypedStruct() {
+    public void should_map_given_nested_typed_struct() {
         TypedStruct struct = TypedStruct.create()
                 .put("field1", TypedStruct.create().put("field2", "value2"));
 
@@ -63,7 +63,7 @@ public void shouldMapGivenNestedTypedStruct() {
 
     @Test
     @SuppressWarnings("unchecked")
-    public void shouldMapGivenTypeStructWithArrayField() {
+    public void should_map_given_type_struct_with_array_field() {
         TypedStruct struct = TypedStruct.create()
                 .put("field1", Collections.singletonList("value"));
 
@@ -78,7 +78,7 @@ public void shouldMapGivenTypeStructWithArrayField() {
 
     @Test
     @SuppressWarnings("unchecked")
-    public void shouldMapGivenTypeStructWithArrayOfStruct() {
+    public void should_Map_given_type_struct_with_array_of_struct() {
         TypedStruct struct = TypedStruct.create()
                 .put("field1", Collections.singletonList(TypedStruct.create().put("field2", "value")));
 
@@ -108,7 +108,7 @@ public void shouldMapGivenTypeStructWithMapField() {
 
     @Test
     @SuppressWarnings("unchecked")
-    public void shouldMapGivenTypeStructWithMapWithStructValue() {
+    public void should_map_given_type_struct_with_map_with_struct_value() {
         TypedStruct struct = TypedStruct.create()
             .put("field1",
                 Collections.singletonMap(
@@ -126,7 +126,7 @@ public void shouldMapGivenTypeStructWithMapWithStructValue() {
     }
 
     @Test
-    public void shouldMapGivenTypeStructWithNullValue() {
+    public void should_map_given_type_struct_with_null_value() {
         TypedStruct struct = TypedStruct.create()
                 .put("field1", "value1")
                 .put("field2", Schema.none(), null);
@@ -139,6 +139,20 @@ public void shouldMapGivenTypeStructWithNullValue() {
         Assert.assertNull(connectStruct.schema().field("field2"));
     }
 
+    @Test
+    public void should_map_given_type_struct_with_empty_array() {
+        TypedStruct struct = TypedStruct.create()
+                .put("field1", "value1")
+                .put("field2", Schema.array(Collections.emptyList(), null), Collections.emptyList());
+
+        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"));
+    }
+
     @Test
     public void test_normalize_schema_name() {
         Assert.assertEquals("Foo", ConnectSchemaMapper.normalizeSchemaName("foo"));

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

@@ -22,6 +22,7 @@
 import com.jsoniter.ValueType;
 import com.jsoniter.any.Any;
 import io.streamthoughts.kafka.connect.filepulse.data.Schema;
+import io.streamthoughts.kafka.connect.filepulse.data.SchemaSupplier;
 import io.streamthoughts.kafka.connect.filepulse.data.Type;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
@@ -147,7 +148,7 @@ public TypedValue read(JsonIterator it) throws IOException {
                 type = read.type();
                 array.add(read.value());
             }
-            return TypedValue.array(array, type);
+            return TypedValue.array(array, type != null ? Schema.of(type) : SchemaSupplier.lazy(array).get());
         }
     }
     private static class ObjectJsonFieldAccessor implements JsonFieldAccessor<TypedStruct> {

connect-file-pulse-filters/src/test/java/io/streamthoughts/kafka/connect/filepulse/filter/JSONFilterTest.java

@@ -19,14 +19,17 @@
 package io.streamthoughts.kafka.connect.filepulse.filter;
 
 import io.streamthoughts.kafka.connect.filepulse.config.JSONFilterConfig;
+import io.streamthoughts.kafka.connect.filepulse.data.SchemaSupplier;
 import io.streamthoughts.kafka.connect.filepulse.data.Type;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
+import io.streamthoughts.kafka.connect.filepulse.source.internal.ConnectSchemaMapper;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -45,6 +48,7 @@ public class JSONFilterTest {
     private static final TypedStruct BYTES_RECORD = create().put("message", JSON.getBytes(StandardCharsets.UTF_8));
 
     private static final TypedStruct ARRAY_STRING_RECORD = create().put("message", ARRAY_STRING_JSON);
+    private static final TypedStruct ARRAY_EMPTY_RECORD = create().put("message", "[]");
     private static final TypedStruct ARRAY_STRUCT_RECORD = create().put("message", ARRAY_STRUCT_JSON);
 
     private JSONFilter filter;
@@ -83,12 +87,28 @@ public void should_add_parsed_array_into_specific_field_given_target_field() {
         }});
 
         List<TypedStruct> expected = Collections.singletonList(
-            create().put("message", ARRAY_STRING_JSON).put("myTarget", TypedValue.array(Arrays.asList("foo", "bar"), Type.STRING))
+            create().put("message", ARRAY_STRING_JSON)
+                    .put("myTarget", TypedValue.array(Arrays.asList("foo", "bar"), Type.STRING))
         );
 
         assertOutput(filter.apply(null, ARRAY_STRING_RECORD, false).collect(), expected);
     }
 
+    @Test
+    public void should_add_parsed_array_into_specific_field_given_target_field_and_empty_array() {
+        filter.configure(new HashMap<String, Object>(){{
+            put(JSONFilterConfig.JSON_TARGET_CONFIG, "myTarget");
+        }});
+
+        List<TypedStruct> expected = Collections.singletonList(
+                create().put("message", "[]")
+                        .put("myTarget", TypedValue.array(new ArrayList<>(), SchemaSupplier.lazy(new ArrayList<>()).get()))
+        );
+
+        final List<TypedStruct> collect = filter.apply(null, ARRAY_EMPTY_RECORD, false).collect();
+        assertOutput(collect, expected);
+    }
+
     @Test
     public void should_add_parsed_array_into_specific_field_given_target_and_explode_true() {
         filter.configure(new HashMap<String, Object>(){{

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

@@ -34,7 +34,7 @@
 
 public class DefaultJSONStructConverterTest {
 
-    private JSONStructConverter converter = new DefaultJSONStructConverter();
+    private final JSONStructConverter converter = new DefaultJSONStructConverter();
 
     @Test
     public void shouldConvertGivenArrayOfComplexType() throws Exception {