feat: extend smt to support key/value/raw types

Author: jeqo

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,15 +43,23 @@ 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) {
+
+    protected abstract Schema operatingSchema(R record);
+
+    protected abstract Object operatingValue(R record);
+
+    private R applyWithSchema(final R record) {
         final Struct struct = (Struct) record.value();
         final Optional<String> fieldValue = extractStructFieldValue(struct, fieldName);
         return filterCondition.test(fieldValue) ? record : null;
@@ -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.isEmpty()) {
+            final Optional<String> value = extractSchemalessFieldValue(operatingValue(record));
+            return filterCondition.test(value) ? record : null;
+        } else {
+            final Map<String, Object> map = (Map<String, Object>) record.value();
+            final Optional<String> fieldValue = extractSchemalessFieldValue(map.get(fieldName));
+            return filterCondition.test(fieldValue) ? record : null;
         }
+    }
 
-        final Object fieldValue = map.get(fieldName);
-
+    private Optional<String> extractSchemalessFieldValue(final Object fieldValue) {
+        if (fieldValue == null) return Optional.empty();
         Optional<String> text = Optional.empty();
         if (isSupportedType(fieldValue)) {
             text = Optional.of(fieldValue.toString());
@@ -150,4 +158,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();
+        }
+    }
 }

src/test/java/io/aiven/kafka/connect/transforms/FilterByFieldValueTest.java

@@ -45,7 +45,7 @@ class FilterByFieldValueTest {
 
     @Test
     void shouldFilterOutRecordsEqualsToReadEvents() {
-        final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue<>();
+        final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue.Value<>();
         filter.configure(Map.of(
                 "field.name", "op",
                 "field.value", "r",
@@ -72,7 +72,7 @@ final var record = new SourceRecord(null, null, "some_topic", VALUE_SCHEMA, valu
 
     @Test
     void shouldKeepRecordsNotEqualsToReadEvents() {
-        final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue<>();
+        final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue.Value<>();
         filter.configure(Map.of(
                 "field.name", "op",
                 "field.value", "r",
@@ -99,7 +99,7 @@ final var record = new SourceRecord(null, null, "some_topic", VALUE_SCHEMA, valu
 
     @Test
     void shouldFilterOutRecordsNotEqualsReadEvents() {
-        final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue<>();
+        final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue.Value<>();
         filter.configure(Map.of(
                 "field.name", "op",
                 "field.value", "r",
@@ -126,7 +126,7 @@ final var record = new SourceRecord(null, null, "some_topic", VALUE_SCHEMA, valu
 
     @Test
     void shouldFilterOutMapFieldValue() {
-        final FilterByFieldValue<SourceRecord> filterByFieldValue = new FilterByFieldValue<>();
+        final FilterByFieldValue<SourceRecord> filterByFieldValue = new FilterByFieldValue.Value<>();
         final Map<String, String> configs = new HashMap<>();
         configs.put("field.name", "language");
         configs.put("field.value.pattern", ".*Java.*");
@@ -137,7 +137,7 @@ void shouldFilterOutMapFieldValue() {
         value.put("name", "John Doe");
         value.put("language", "Java");
 
-        final var record = new SourceRecord(null, null, "some_topic", Schema.STRING_SCHEMA, value);
+        final var record = new SourceRecord(null, null, "some_topic", null, value);
 
         final var actual = filterByFieldValue.apply(record);
         assertNull(actual, "The record should be filtered out");