feat(filters): add support for dot-notation to SplitFilter

Author: fhussonnois

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/SplitFilter.java

@@ -59,11 +59,11 @@ public RecordsIterable<TypedStruct> apply(final FilterContext context,
 
         for (final String key : configs.split()) {
 
-            if (!record.has(key)) {
+            if (!record.exists(key)) {
                 throw new FilterException("Invalid field name '" + key + "'");
             }
 
-            TypedValue value = record.get(key);
+            TypedValue value = record.find(key);
             if (value.type() != Type.STRING) {
                 throw new FilterException("Cannot split field '" + key + "' of type '" + value.type() + "'");
             }

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

@@ -41,21 +41,38 @@ public void setUp() {
     }
 
     @Test
-    public void shouldSplitGivenExistingField() {
+    public void should_split_given_existing_field() {
 
         configs.put(SplitFilterConfig.MUTATE_SPLIT_CONFIG, "foo");
         filter.configure(configs);
 
         TypedStruct record = TypedStruct.create().put("foo", "val0,val1,val2");
-        List<TypedStruct> results = this.filter.apply(null, record, false).collect();
+        List<TypedStruct> results = filter.apply(null, record, false).collect();
+
+        assertOutput(results, "foo");
+    }
+
+    @Test
+    public void should_split_given_existing_path() {
+
+        configs.put(SplitFilterConfig.MUTATE_SPLIT_CONFIG, "foo.bar");
+        filter.configure(configs);
+
+        TypedStruct record = TypedStruct.create().insert("foo.bar", "val0,val1,val2");
+        List<TypedStruct> results = filter.apply(null, record, false).collect();
 
         Assert.assertNotNull(results);
         Assert.assertEquals(1, results.size());
+        assertOutput(results, "foo.bar");
+    }
 
+    private void assertOutput(final List<TypedStruct> results, final String field) {
+        Assert.assertNotNull(results);
+        Assert.assertEquals(1, results.size());
         TypedStruct result = results.get(0);
-        List<String> array = result.getArray("foo");
+        List<String> array = result.getArray(field);
         Assert.assertEquals(3, array.size());
-        for (int i = 0 ; i < 3; i++) {
+        for (int i = 0; i < 3; i++) {
             Assert.assertEquals("val" + i, array.get(i));
         }
     }