fix(filters): Fix AppendFilter should support substitution expression for field config

Author: fhussonnois

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

@@ -20,6 +20,7 @@
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.expression.Expression;
 import io.streamthoughts.kafka.connect.filepulse.expression.StandardEvaluationContext;
+import io.streamthoughts.kafka.connect.filepulse.expression.ValueExpression;
 import io.streamthoughts.kafka.connect.filepulse.expression.parser.regex.RegexExpressionParser;
 import io.streamthoughts.kafka.connect.filepulse.reader.RecordsIterable;
 import org.apache.kafka.common.config.ConfigDef;
@@ -36,10 +37,12 @@ public class AppendFilter extends AbstractMergeRecordFilter<AppendFilter> {
     private AppendFilterConfig config;
 
     private List<Expression> values;
-    private Expression target;
+    private Expression fieldExpression;
 
     private RegexExpressionParser parser;
 
+    private boolean mustEvaluateWriteExpression = true;
+
     /**
      * {@inheritDoc}
      */
@@ -50,8 +53,15 @@ public void configure(final Map<String, ?> props) {
 
         parser = new RegexExpressionParser();
         // currently, multiple expressions is not supported
-        values = Collections.singletonList(parser.parseExpression(config.value(), DEFAULT_ROOT_OBJECT, true));
-        target = parser.parseExpression(config.field());
+        values = Collections.singletonList(parser.parseExpression(config.value(), DEFAULT_ROOT_OBJECT));
+
+        fieldExpression = parser.parseExpression(config.field(), DEFAULT_ROOT_OBJECT);
+
+        // Check whether the field expression can be pre-evaluated
+        if (fieldExpression instanceof ValueExpression) {
+            fieldExpression = evaluateWriteExpression(new StandardEvaluationContext(new Object()));
+            mustEvaluateWriteExpression = false;
+        }
     }
 
     /**
@@ -70,34 +80,42 @@ protected RecordsIterable<TypedStruct> apply(final FilterContext context,
                                                  final TypedStruct record) throws FilterException {
 
         InternalFilterContext internalContext = (InternalFilterContext) context;
+        internalContext.setValue(record);
 
         StandardEvaluationContext readEvaluationContext = new StandardEvaluationContext(
                 internalContext,
                 internalContext.variables()
         );
 
-        final String evaluatedTarget = target.readValue(readEvaluationContext, String.class);
-        final Expression targetExpression = parser.parseExpression(evaluatedTarget, DEFAULT_ROOT_OBJECT, false);
+        final Expression writeExpression = evaluateWriteExpression(readEvaluationContext);
 
         final TypedStruct target = TypedStruct.struct();
         for (final Expression expression : values) {
 
             internalContext.setValue(record);
-            final Object typed = expression.readValue(readEvaluationContext);
+            final Object value = expression.readValue(readEvaluationContext);
 
-            if (typed != null) {
+            if (value != null) {
                 internalContext.setValue(target);
 
                 final StandardEvaluationContext writeEvaluationContext = new StandardEvaluationContext(
                     internalContext,
                     internalContext.variables()
                 );
-                targetExpression.writeValue(typed, writeEvaluationContext);
+                writeExpression.writeValue(value, writeEvaluationContext);
             }
         }
         return RecordsIterable.of(target);
     }
 
+    private Expression evaluateWriteExpression(final StandardEvaluationContext evaluationContext) {
+        if (mustEvaluateWriteExpression) {
+            final String evaluated = fieldExpression.readValue(evaluationContext, String.class);
+            return parser.parseExpression(evaluated, DEFAULT_ROOT_OBJECT, false);
+        }
+        return fieldExpression;
+    }
+
     /**
      * {@inheritDoc}
      */

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

@@ -31,6 +31,9 @@
 
 public class AppendFilterTest {
 
+    private final static TypedStruct STRUCT =  new TypedStruct().put("values", Arrays.asList("foo", "bar"));
+    private final static String EXPRESSION = "{{ extract_array(values,0) }}-{{ extract_array(values,1) }}";
+
     private AppendFilter filter;
 
     private FilterContext context;
@@ -48,16 +51,38 @@ public void setUp() {
     }
 
     @Test
-    public void testGivenSubstitutionExpression() {
+    public void shouldSupportSubstitutionExpressionForValueConfig() {
         configs.put(AppendFilterConfig.APPEND_FIELD_CONFIG, "target");
-        configs.put(AppendFilterConfig.APPEND_VALUE_CONFIG, "{{ extract_array(values,0) }}-{{ extract_array(values,1) }}");
+        configs.put(AppendFilterConfig.APPEND_VALUE_CONFIG, EXPRESSION);
+        filter.configure(configs);
+
+        RecordsIterable<TypedStruct> output = filter.apply(context, STRUCT);
+        Assert.assertNotNull(output);
+        TypedStruct result = output.collect().get(0);
+        Assert.assertEquals("foo-bar", result.getString("target"));
+    }
+
+    @Test
+    public void shouldSupportPropertyExpressionForFieldConfig() {
+        configs.put(AppendFilterConfig.APPEND_FIELD_CONFIG, "$value.target");
+        configs.put(AppendFilterConfig.APPEND_VALUE_CONFIG, EXPRESSION);
         filter.configure(configs);
 
-        final TypedStruct struct = new TypedStruct();
-        struct.put("values", Arrays.asList("foo", "bar"));
-        RecordsIterable<TypedStruct> output = filter.apply(context, struct);
+        RecordsIterable<TypedStruct> output = filter.apply(context, STRUCT);
         Assert.assertNotNull(output);
         TypedStruct result = output.collect().get(0);
         Assert.assertEquals("foo-bar", result.getString("target"));
     }
+
+    @Test
+    public void shouldSupportSubstitutionExpressionForFieldConfig() {
+        configs.put(AppendFilterConfig.APPEND_FIELD_CONFIG, "{{ extract_array(values,0) }}");
+        configs.put(AppendFilterConfig.APPEND_VALUE_CONFIG, EXPRESSION);
+        filter.configure(configs);
+
+        RecordsIterable<TypedStruct> output = filter.apply(context, STRUCT);
+        Assert.assertNotNull(output);
+        TypedStruct result = output.collect().get(0);
+        Assert.assertEquals("foo-bar", result.getString("foo"));
+    }
 }