fix(filters): fix AppendFilter overwrite propery not working with ScEL (#62)

Resolves: #62

Author: fhussonnois

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/AppendFilterConfig.java

@@ -18,6 +18,8 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.config;
 
+import io.streamthoughts.kafka.connect.filepulse.expression.Expression;
+import io.streamthoughts.kafka.connect.filepulse.expression.parser.ExpressionParsers;
 import org.apache.kafka.common.config.ConfigDef;
 
 import java.util.Map;
@@ -42,8 +44,8 @@ public AppendFilterConfig(final Map<?, ?> originals) {
         super(configDef(), originals);
     }
 
-    public String field() {
-        return getString(APPEND_FIELD_CONFIG);
+    public Expression field() {
+        return ExpressionParsers.parseExpression(getString(APPEND_FIELD_CONFIG));
     }
 
     public boolean overwrite() {

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

@@ -21,6 +21,7 @@
 import io.streamthoughts.kafka.connect.filepulse.config.AppendFilterConfig;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.expression.Expression;
+import io.streamthoughts.kafka.connect.filepulse.expression.PropertyExpression;
 import io.streamthoughts.kafka.connect.filepulse.expression.StandardEvaluationContext;
 import io.streamthoughts.kafka.connect.filepulse.expression.parser.ExpressionParsers;
 import io.streamthoughts.kafka.connect.filepulse.reader.RecordsIterable;
@@ -38,6 +39,8 @@ public class AppendFilter extends AbstractMergeRecordFilter<AppendFilter> {
     private List<Expression> values;
     private Expression fieldExpression;
 
+    protected String target;
+
     /**
      * {@inheritDoc}
      */
@@ -49,7 +52,7 @@ public void configure(final Map<String, ?> props) {
         // currently, multiple expressions is not supported
         values = Collections.singletonList(ExpressionParsers.parseExpression(config.value()));
 
-        fieldExpression = ExpressionParsers.parseExpression(config.field());
+        fieldExpression = config.field();
     }
 
     /**
@@ -97,15 +100,18 @@ protected RecordsIterable<TypedStruct> apply(final FilterContext context,
     }
 
     private Expression mayEvaluateWriteExpression(final StandardEvaluationContext evaluationContext) {
+        Expression expression = fieldExpression;
         if (!fieldExpression.canWrite()) {
             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 ExpressionParsers.parseExpression(evaluated);
+            expression = ExpressionParsers.parseExpression(evaluated);
         }
-        return fieldExpression;
+
+        target = ((PropertyExpression)expression).getAttribute();
+        return expression;
     }
 
     /**
@@ -114,7 +120,7 @@ private Expression mayEvaluateWriteExpression(final StandardEvaluationContext ev
     @Override
     protected Set<String> overwrite() {
         if (config.overwrite()) {
-            return Collections.singleton(config.field());
+            return Collections.singleton(target);
         }
         return Collections.emptySet();
     }

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

@@ -96,4 +96,26 @@ public void shouldSupportPropertyExpressionWithScopeForFieldConfig() {
         filter.apply(context, STRUCT);
         Assert.assertEquals("my-topic-foo", context.topic());
     }
+
+    @Test
+    public void shouldOverwriteExistingValueGivenOverwriteTrue() {
+        configs.put(AppendFilterConfig.APPEND_FIELD_CONFIG, "$value.field");
+        configs.put(AppendFilterConfig.APPEND_VALUE_CONFIG, "bar");
+        configs.put(AppendFilterConfig.APPEND_OVERWRITE_CONFIG, "true");
+        filter.configure(configs);
+        final TypedStruct input = TypedStruct.create().put("field", "foo");
+        RecordsIterable<TypedStruct> results = filter.apply(context, input, false);
+        Assert.assertEquals("bar", results.last().getString("field"));
+    }
+
+    @Test
+    public void shouldMergeExistingValueGivenOverwriteFalse() {
+        configs.put(AppendFilterConfig.APPEND_FIELD_CONFIG, "$value.field");
+        configs.put(AppendFilterConfig.APPEND_VALUE_CONFIG, "bar");
+        configs.put(AppendFilterConfig.APPEND_OVERWRITE_CONFIG, "false");
+        filter.configure(configs);
+        final TypedStruct input = TypedStruct.create().put("field", "foo");
+        RecordsIterable<TypedStruct> results = filter.apply(context, input, false);
+        Assert.assertEquals("[foo, bar]", results.last().getArray("field").toString());
+    }
 }