Merge pull request #84 from aiven/jeqo-refactor-82

Author: jeqo

README.md

@@ -143,7 +143,7 @@ transforms.MakeTombstone.type=io.aiven.kafka.connect.transforms.MakeTombstone
 
 ### `FilterByFieldValue`
 
-This transformation allows filtering records based on a specific field value or a given regex pattern.
+This transformation allows filtering records based either on a specific field or whole value and a matching expected value or regex pattern.
 
 Here is an example of this transformation configuration:
 
@@ -156,8 +156,12 @@ transforms.Filter.field.value.pattern=<regex_pattern>
 transforms.Filter.field.value.matches=<true|false>
 ```
 
+If `field.name` is empty, the whole value is considered for filtering.
+
 Either `field.value` or `field.value.pattern` must be defined to apply filter.
 
+Only, `string`, `numeric` and `boolean` types are considered for matching purposes, other types are ignored.
+
 ## License
 
 This project is licensed under the [Apache License, Version 2.0](LICENSE).

src/main/java/io/aiven/kafka/connect/transforms/FilterByFieldValue.java

@@ -34,7 +34,7 @@
 
 import static org.apache.kafka.connect.data.Schema.Type.STRING;
 
-public class FilterByFieldValue<R extends ConnectRecord<R>> implements Transformation<R> {
+public abstract class FilterByFieldValue<R extends ConnectRecord<R>> implements Transformation<R> {
 
     private String fieldName;
     private Optional<String> fieldExpectedValue;
@@ -43,21 +43,29 @@ public class FilterByFieldValue<R extends ConnectRecord<R>> implements Transform
 
     @Override
     public R apply(final R record) {
-        if (record.value() instanceof Struct) {
-            return handleStruct(record);
-        } else if (record.value() instanceof Map) {
-            return handleMap(record);
+        if (operatingValue(record) == null) {
+            return record;
+        }
+
+        if (operatingSchema(record) == null) {
+            return applySchemaless(record);
+        } else {
+            return applyWithSchema(record);
         }
-        return record; // if record is other than map or struct, pass-by
     }
 
-    private R handleStruct(final R record) {
-        final Struct struct = (Struct) record.value();
-        final Optional<String> fieldValue = extractStructFieldValue(struct, fieldName);
+
+    protected abstract Schema operatingSchema(R record);
+
+    protected abstract Object operatingValue(R record);
+
+    private R applyWithSchema(final R record) {
+        final Struct struct = (Struct) operatingValue(record);
+        final Optional<String> fieldValue = getStructFieldValue(struct, fieldName);
         return filterCondition.test(fieldValue) ? record : null;
     }
 
-    private Optional<String> extractStructFieldValue(final Struct struct, final String fieldName) {
+    private Optional<String> getStructFieldValue(final Struct struct, final String fieldName) {
         final Schema schema = struct.schema();
         final Field field = schema.field(fieldName);
         final Object fieldValue = struct.get(field);
@@ -72,19 +80,19 @@ private Optional<String> extractStructFieldValue(final Struct struct, final Stri
     }
 
     @SuppressWarnings("unchecked")
-    private R handleMap(final R record) {
-        final Map<String, Object> map = (Map<String, Object>) record.value();
-        final Optional<String> fieldValue = extractMapFieldValue(map, fieldName);
-        return filterCondition.test(fieldValue) ? record : null;
-    }
-
-    private Optional<String> extractMapFieldValue(final Map<String, Object> map, final String fieldName) {
-        if (!map.containsKey(fieldName)) {
-            return Optional.empty();
+    private R applySchemaless(final R record) {
+        if (fieldName == null || fieldName.isEmpty()) {
+            final Optional<String> value = getSchemalessFieldValue(operatingValue(record));
+            return filterCondition.test(value) ? record : null;
+        } else {
+            final Map<String, Object> map = (Map<String, Object>) operatingValue(record);
+            final Optional<String> fieldValue = getSchemalessFieldValue(map.get(fieldName));
+            return filterCondition.test(fieldValue) ? record : null;
         }
+    }
 
-        final Object fieldValue = map.get(fieldName);
-
+    private Optional<String> getSchemalessFieldValue(final Object fieldValue) {
+        if (fieldValue == null) return Optional.empty();
         Optional<String> text = Optional.empty();
         if (isSupportedType(fieldValue)) {
             text = Optional.of(fieldValue.toString());
@@ -108,18 +116,24 @@ public ConfigDef config() {
         return new ConfigDef()
                 .define("field.name",
                         ConfigDef.Type.STRING,
+                        null,
                         ConfigDef.Importance.HIGH,
-                        "The field name to filter by")
+                        "The field name to filter by." +
+                                "Schema-based records (Avro), schemaless (e.g. JSON), and raw values are supported." +
+                                "If empty, the whole key/value record will be filtered.")
                 .define("field.value",
-                        ConfigDef.Type.STRING, null,
+                        ConfigDef.Type.STRING,
+                        null,
                         ConfigDef.Importance.HIGH,
                         "Expected value to match. Either define this, or a regex pattern")
                 .define("field.value.pattern",
-                        ConfigDef.Type.STRING, null,
+                        ConfigDef.Type.STRING,
+                        null,
                         ConfigDef.Importance.HIGH,
                         "The pattern to match. Either define this, or an expected value")
                 .define("field.value.matches",
-                        ConfigDef.Type.BOOLEAN, true,
+                        ConfigDef.Type.BOOLEAN,
+                        true,
                         ConfigDef.Importance.MEDIUM,
                         "The filter mode, 'true' for matching or 'false' for non-matching");
     }
@@ -150,4 +164,31 @@ public void configure(final Map<String, ?> configs) {
                 ? matchCondition
                 : (result -> !matchCondition.test(result));
     }
+
+
+    public static final class Key<R extends ConnectRecord<R>> extends FilterByFieldValue<R> {
+
+        @Override
+        protected Schema operatingSchema(R record) {
+            return record.keySchema();
+        }
+
+        @Override
+        protected Object operatingValue(R record) {
+            return record.key();
+        }
+    }
+
+    public static final class Value<R extends ConnectRecord<R>> extends FilterByFieldValue<R> {
+
+        @Override
+        protected Schema operatingSchema(R record) {
+            return record.valueSchema();
+        }
+
+        @Override
+        protected Object operatingValue(R record) {
+            return record.value();
+        }
+    }
 }