fix(api): fix ReflectionPropertyAccessor should convert value to expected type (#59)

Additional changes:
 - AppendFilter throws a FilterException if expression value for target
field return null

Resolves: GH-59

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/expression/accessor/ReflectivePropertyAccessor.java

@@ -19,13 +19,14 @@
 package io.streamthoughts.kafka.connect.filepulse.expression.accessor;
 
 import io.streamthoughts.kafka.connect.filepulse.expression.EvaluationContext;
+import io.streamthoughts.kafka.connect.filepulse.expression.converter.Converters;
 
 import java.lang.reflect.Method;
 import java.util.Objects;
 
 public class ReflectivePropertyAccessor implements PropertyAccessor {
 
-    private static final String GETTER_PREFIX = "of";
+    private static final String GETTER_PREFIX = "get";
     private static final String SETTER_PREFIX = "set";
 
     /**
@@ -83,7 +84,9 @@ public void write(final EvaluationContext context,
             Method method = findSetterMethodForProperty(type, name);
             if (method != null ) {
                 method.setAccessible(true);
-                method.invoke(target, newValue);
+                Class<?> expectedSetterType = method.getParameterTypes()[0];
+                Object converted = Converters.converts(context.getPropertyConverter(), newValue, expectedSetterType);
+                method.invoke(target, converted);
                 return;
             }
         } catch (Exception e) {

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/expression/parser/regex/RegexExpressionParser.java

@@ -101,7 +101,7 @@ public Expression parseExpression(final String expression,
                 }
             }
         } else {
-            compiledExpression = matchers.matches(expression, defaultRootObject, substitution);
+            compiledExpression = matchers.matches(expression, defaultRootObject, false);
         }
 
         if (compiledExpression == null) {

connect-file-pulse-api/src/test/java/io/streamthoughts/kafka/connect/filepulse/expression/accessor/ReflectivePropertyAccessorTest.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2019-2020 StreamThoughts.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.streamthoughts.kafka.connect.filepulse.expression.accessor;
+
+import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
+import io.streamthoughts.kafka.connect.filepulse.expression.StandardEvaluationContext;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ReflectivePropertyAccessorTest {
+
+    private StandardEvaluationContext context = new StandardEvaluationContext(new Object());
+
+
+    @Test(expected = AccessException.class)
+    public void should_thrown_when_writing_invalid_property_using_getter_method_given_pojo() {
+        ReflectivePropertyAccessor accessor = new ReflectivePropertyAccessor();
+        accessor.write(context, new DummyObject("foo"), "unknown", "");
+    }
+
+    @Test(expected = AccessException.class)
+    public void should_thrown_when_reading_invalid_property_using_getter_method_given_pojo() {
+        ReflectivePropertyAccessor accessor = new ReflectivePropertyAccessor();
+        accessor.read(context, new DummyObject("foo"), "unknown");
+    }
+
+    @Test
+    public void should_write_property_using_setter_method_given_pojo_and_expected_parameter() {
+        ReflectivePropertyAccessor accessor = new ReflectivePropertyAccessor();
+        DummyObject object = new DummyObject(null);
+        accessor.write(context, object, "field", "foo");
+        Assert.assertEquals("foo", object.field);
+    }
+
+    @Test
+    public void should_write_null_property_using_setter_method_given_pojo_and_expected_parameter() {
+        ReflectivePropertyAccessor accessor = new ReflectivePropertyAccessor();
+        DummyObject object = new DummyObject("foo");
+        accessor.write(context, object, "field", null);
+        Assert.assertEquals(null, object.field);
+    }
+
+    @Test
+    public void should_write_property_using_setter_method_given_pojo() {
+        ReflectivePropertyAccessor accessor = new ReflectivePropertyAccessor();
+        DummyObject object = new DummyObject(null);
+        accessor.write(context, object, "field", TypedValue.string("foo"));
+        Assert.assertEquals("foo", object.field);
+    }
+
+    @Test
+    public void should_read_property_using_getter_method_given_pojo() {
+        ReflectivePropertyAccessor accessor = new ReflectivePropertyAccessor();
+        Object object = accessor.read(context, new DummyObject("foo"), "field");
+        Assert.assertEquals("foo", object);
+    }
+
+    public static class DummyObject {
+
+        private String field;
+
+        DummyObject(String myField) {
+            this.field = myField;
+        }
+
+        public String getField() {
+            return field;
+        }
+
+        public void setField(String field) {
+            this.field = field;
+        }
+    }
+
+}

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

@@ -113,6 +113,10 @@ protected RecordsIterable<TypedStruct> apply(final FilterContext context,
     private Expression evaluateWriteExpression(final StandardEvaluationContext evaluationContext) {
         if (mustEvaluateWriteExpression) {
             final String evaluated = fieldExpression.readValue(evaluationContext, String.class);
+            if (evaluated == null) {
+                throw new FilterException("Invalid value for property 'field'. Evaluation of expression '"
+                    + fieldExpression.originalExpression() + " 'returns 'null'.");
+            }
             return parser.parseExpression(evaluated, DEFAULT_ROOT_OBJECT, false);
         }
         return fieldExpression;

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

@@ -87,4 +87,13 @@ public void shouldSupportSubstitutionExpressionForFieldConfig() {
         TypedStruct result = output.collect().get(0);
         Assert.assertEquals("foo-bar", result.getString("foo"));
     }
+
+    @Test
+    public void shouldSupportPropertyExpressionWithScopeForFieldConfig() {
+        configs.put(AppendFilterConfig.APPEND_FIELD_CONFIG, "$topic");
+        configs.put(AppendFilterConfig.APPEND_VALUE_CONFIG, "my-topic-{{ extract_array(values,0) }}");
+        filter.configure(configs);
+        filter.apply(context, STRUCT);
+        Assert.assertEquals("my-topic-foo", context.topic());
+    }
 }