feat: use Values to parse value_and_schemas

jeqo · jeqo · commit 588d53db2da7 · 2023-03-30T10:23:19.000+03:00
diff --git a/src/main/java/io/aiven/kafka/connect/transforms/FilterByFieldValue.java b/src/main/java/io/aiven/kafka/connect/transforms/FilterByFieldValue.java
@@ -20,12 +20,14 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
+import java.util.regex.Pattern;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.connector.ConnectRecord;
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.data.Values;
 import org.apache.kafka.connect.transforms.Transformation;
@@ -35,7 +37,68 @@ public abstract class FilterByFieldValue<R extends ConnectRecord<R>> implements
     private String fieldName;
     private Optional<String> fieldExpectedValue;
     private Optional<String> fieldValuePattern;
-    private Predicate<String> filterCondition;
+
+    @Override
+    public ConfigDef config() {
+        return new ConfigDef()
+            .define("field.name",
+                ConfigDef.Type.STRING,
+                null,
+                ConfigDef.Importance.HIGH,
+                "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.Importance.HIGH,
+                "Expected value to match. Either define this, or a regex pattern")
+            .define("field.value.pattern",
+                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.Importance.MEDIUM,
+                "The filter mode, 'true' for matching or 'false' for non-matching");
+    }
+
+    @Override
+    public void configure(final Map<String, ?> configs) {
+        final AbstractConfig config = new AbstractConfig(config(), configs);
+        this.fieldName = config.getString("field.name");
+        this.fieldExpectedValue = Optional.ofNullable(config.getString("field.value"));
+        this.fieldValuePattern = Optional.ofNullable(config.getString("field.value.pattern"));
+        final boolean expectedValuePresent = fieldExpectedValue.isPresent();
+        final boolean regexPatternPresent = fieldValuePattern.map(s -> !s.isEmpty()).orElse(false);
+        if (expectedValuePresent == regexPatternPresent) {
+            throw new ConfigException(
+                "Either field.value or field.value.pattern have to be set to apply filter transform");
+        }
+        final Predicate<SchemaAndValue> matchCondition;
+
+        if (expectedValuePresent) {
+            final SchemaAndValue expectedSchemaAndValue = Values.parseString(fieldExpectedValue.get());
+            matchCondition = schemaAndValue -> expectedSchemaAndValue.value().equals(schemaAndValue.value());
+        } else {
+            final String pattern = fieldValuePattern.get();
+            final Predicate<String> regexPredicate = Pattern.compile(pattern).asPredicate();
+            matchCondition = schemaAndValue ->
+                schemaAndValue != null
+                    && regexPredicate.test(Values.convertToString(schemaAndValue.schema(), schemaAndValue.value()));
+        }
+
+        this.filterCondition = config.getBoolean("field.value.matches")
+            ? matchCondition
+            : (result -> !matchCondition.test(result));
+    }
+    private Predicate<SchemaAndValue> filterCondition;
+
+    protected abstract Schema operatingSchema(R record);
+
+    protected abstract Object operatingValue(R record);
 
     @Override
     public R apply(final R record) {
@@ -50,102 +113,46 @@ public R apply(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) operatingValue(record);
-        final String fieldValue = getStructFieldValue(struct, fieldName).orElse(null);
-        return filterCondition.test(fieldValue) ? record : null;
+        final SchemaAndValue schemaAndValue = getStructFieldValue(struct, fieldName).orElse(null);
+        return filterCondition.test(schemaAndValue) ? record : null;
     }
 
-    private Optional<String> getStructFieldValue(final Struct struct, final String fieldName) {
+    private Optional<SchemaAndValue> getStructFieldValue(final Struct struct, final String fieldName) {
         final Schema schema = struct.schema();
         final Field field = schema.field(fieldName);
         final Object fieldValue = struct.get(field);
-        return Optional.ofNullable(Values.convertToString(field.schema(), fieldValue));
+        if (fieldValue == null) {
+            return Optional.empty();
+        } else {
+            return Optional.of(new SchemaAndValue(field.schema(), struct.get(field)));
+        }
     }
 
     @SuppressWarnings("unchecked")
     private R applySchemaless(final R record) {
         if (fieldName == null || fieldName.isEmpty()) {
-            final String value = getSchemalessFieldValue(operatingValue(record)).orElse(null);
-            return filterCondition.test(value) ? record : null;
+            final SchemaAndValue schemaAndValue = getSchemalessFieldValue(operatingValue(record)).orElse(null);
+            return filterCondition.test(schemaAndValue) ? record : null;
         } else {
             final Map<String, Object> map = (Map<String, Object>) operatingValue(record);
-            final String fieldValue = getSchemalessFieldValue(map.get(fieldName)).orElse(null);
-            return filterCondition.test(fieldValue) ? record : null;
+            final SchemaAndValue schemaAndValue = getSchemalessFieldValue(map.get(fieldName)).orElse(null);
+            return filterCondition.test(schemaAndValue) ? record : null;
         }
     }
 
-    private Optional<String> getSchemalessFieldValue(final Object fieldValue) {
+    private Optional<SchemaAndValue> getSchemalessFieldValue(final Object fieldValue) {
         if (fieldValue == null) {
             return Optional.empty();
         }
-        return Optional.ofNullable(Values.convertToString(null, fieldValue));
-    }
-
-    @Override
-    public ConfigDef config() {
-        return new ConfigDef()
-                .define("field.name",
-                        ConfigDef.Type.STRING,
-                        null,
-                        ConfigDef.Importance.HIGH,
-                        "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.Importance.HIGH,
-                        "Expected value to match. Either define this, or a regex pattern")
-                .define("field.value.pattern",
-                        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.Importance.MEDIUM,
-                        "The filter mode, 'true' for matching or 'false' for non-matching");
+        return Optional.of(new SchemaAndValue(Values.inferSchema(fieldValue), fieldValue));
     }
 
     @Override
     public void close() {
     }
 
-    @Override
-    public void configure(final Map<String, ?> configs) {
-        final AbstractConfig config = new AbstractConfig(config(), configs);
-        this.fieldName = config.getString("field.name");
-        this.fieldExpectedValue = Optional.ofNullable(config.getString("field.value"));
-        this.fieldValuePattern = Optional.ofNullable(config.getString("field.value.pattern"));
-        final boolean expectedValuePresent = fieldExpectedValue.map(s -> !s.isEmpty()).orElse(false);
-        final boolean regexPatternPresent = fieldValuePattern.map(s -> !s.isEmpty()).orElse(false);
-        if (expectedValuePresent == regexPatternPresent) {
-            throw new ConfigException(
-                    "Either field.value or field.value.pattern have to be set to apply filter transform");
-        }
-        final Predicate<String> matchCondition;
-
-        if (expectedValuePresent) {
-            final String expectedValue = fieldExpectedValue.get();
-            matchCondition = fieldValue -> fieldValue != null && fieldValue.equals(expectedValue);
-        } else {
-            final String pattern = fieldValuePattern.get();
-            matchCondition = fieldValue -> fieldValue != null && fieldValue.matches(pattern);
-        }
-
-        this.filterCondition = config.getBoolean("field.value.matches")
-            ? matchCondition
-            : (result -> !matchCondition.test(result));
-    }
-
-
     public static final class Key<R extends ConnectRecord<R>> extends FilterByFieldValue<R> {
 
         @Override
diff --git a/src/test/java/io/aiven/kafka/connect/transforms/FilterByFieldValueTest.java b/src/test/java/io/aiven/kafka/connect/transforms/FilterByFieldValueTest.java
@@ -65,19 +65,19 @@ void shouldFilterOutKeyRecordsEqualsToId() {
         final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue.Key<>();
         filter.configure(Map.of(
             "field.name", "id",
-            "field.value", "123",
+            "field.value", "A123",
             "field.value.matches", "false"
         ));
 
         assertNull(filter.apply(prepareStructRecord(
-            struct -> struct.put("id", "123"),
+            struct -> struct.put("id", "A123"),
             struct -> {
-            })), "Record with id '123' should be filtered out");
+            })), "Record with id 'A132' should be filtered out");
         final SourceRecord record = prepareStructRecord(
-            struct -> struct.put("id", "111"),
+            struct -> struct.put("id", "A111"),
             struct -> {
             });
-        assertEquals(record, filter.apply(record), "Record with id not '123' should not be filtered out");
+        assertEquals(record, filter.apply(record), "Record with id not 'A132' should not be filtered out");
     }
 
     @Test
@@ -110,7 +110,7 @@ void shouldFilterOutKeyRecordsNotEqualsToId() {
         final FilterByFieldValue<SourceRecord> filter = new FilterByFieldValue.Key<>();
         filter.configure(Map.of(
             "field.name", "id",
-            "field.value", "123",
+            "field.value", "A123",
             "field.value.matches", "true"
         ));
 
@@ -121,13 +121,13 @@ void shouldFilterOutKeyRecordsNotEqualsToId() {
                     struct -> {
                     }
                 )),
-            "Record with id not equal to '123' should be filtered out");
+            "Record with id not equal to 'A132' should be filtered out");
         final SourceRecord record = prepareStructRecord(
-            struct -> struct.put("id", "123"),
+            struct -> struct.put("id", "A123"),
             struct -> {
             }
         );
-        assertEquals(record, filter.apply(record), "Record with id equal to '123' should not be filtered out");
+        assertEquals(record, filter.apply(record), "Record with id equal to 'A132' should not be filtered out");
     }
 
     @Test
@@ -139,9 +139,9 @@ void shouldFilterOutMapValueRecordsWithRegex() {
         configs.put("field.value.matches", "false");
         filterByFieldValue.configure(configs);
 
-        assertNull(filterByFieldValue.apply(prepareRecord(() -> "42", () -> Map.of("language", "Javascript"))),
+        assertNull(filterByFieldValue.apply(prepareRecord(() -> "A42", () -> Map.of("language", "Javascript"))),
             "The record should be filtered out");
-        final SourceRecord record = prepareRecord(() -> "42", () -> Map.of("language", "Rust"));
+        final SourceRecord record = prepareRecord(() -> "A42", () -> Map.of("language", "Rust"));
         assertEquals(record, filterByFieldValue.apply(record), "The record should not be filtered out");
     }
 
@@ -154,21 +154,21 @@ void shouldFilterOutMapKeyRecordsWithRegex() {
         configs.put("field.value.matches", "false");
         filterByFieldValue.configure(configs);
 
-        assertNull(filterByFieldValue.apply(prepareRecord(() -> Map.of("language", "Javascript"), () -> "42")),
+        assertNull(filterByFieldValue.apply(prepareRecord(() -> Map.of("language", "Javascript"), () -> "A42")),
             "The record should be filtered out");
-        final SourceRecord record = prepareRecord(() -> Map.of("language", "Rust"), () -> "42");
+        final SourceRecord record = prepareRecord(() -> Map.of("language", "Rust"), () -> "A42");
         assertEquals(record, filterByFieldValue.apply(record), "The record should not be filtered out");
     }
 
     @Test
     void shouldFilterOutRawKeyRecords() {
         final FilterByFieldValue<SourceRecord> filterByFieldValue = new FilterByFieldValue.Key<>();
         final Map<String, String> configs = new HashMap<>();
-        configs.put("field.value", "42");
+        configs.put("field.value", "A42");
         configs.put("field.value.matches", "false");
         filterByFieldValue.configure(configs);
 
-        assertNull(filterByFieldValue.apply(prepareRecord(() -> "42", () -> Map.of("language", "Javascript"))),
+        assertNull(filterByFieldValue.apply(prepareRecord(() -> "A42", () -> Map.of("language", "Javascript"))),
             "The record should be filtered out");
         final SourceRecord record = prepareRecord(() -> "43", () -> Map.of("language", "Rust"));
         assertEquals(record, filterByFieldValue.apply(record), "The record should be filtered out");
@@ -178,16 +178,30 @@ void shouldFilterOutRawKeyRecords() {
     void shouldFilterOutRawValueRecords() {
         final FilterByFieldValue<SourceRecord> filterByFieldValue = new FilterByFieldValue.Value<>();
         final Map<String, String> configs = new HashMap<>();
-        configs.put("field.value", "42");
+        configs.put("field.value", "A42");
         configs.put("field.value.matches", "false");
         filterByFieldValue.configure(configs);
 
-        assertNull(filterByFieldValue.apply(prepareRecord(() -> Map.of("language", "Javascript"), () -> "42")),
+        assertNull(filterByFieldValue.apply(prepareRecord(() -> Map.of("language", "Javascript"), () -> "A42")),
             "The record should be filtered out");
         final SourceRecord record = prepareRecord(() -> Map.of("language", "Rust"), () -> "43");
         assertEquals(record, filterByFieldValue.apply(record), "The record should be filtered out");
     }
 
+    @Test
+    void shouldFilterOutRawNumericValueRecords() {
+        final FilterByFieldValue<SourceRecord> filterByFieldValue = new FilterByFieldValue.Value<>();
+        final Map<String, String> configs = new HashMap<>();
+        configs.put("field.value", "42");
+        configs.put("field.value.matches", "false");
+        filterByFieldValue.configure(configs);
+
+        assertNull(filterByFieldValue.apply(prepareRecord(() -> Map.of("language", "Javascript"), () -> (byte) 42)),
+            "The record should be filtered out");
+        final SourceRecord record = prepareRecord(() -> Map.of("language", "Rust"), () -> (byte) 43);
+        assertEquals(record, filterByFieldValue.apply(record), "The record should be filtered out");
+    }
+
     private SourceRecord prepareRecord(
         final Supplier<Object> keySupplier,
         final Supplier<Object> valueSupplier
@@ -217,7 +231,7 @@ private SourceRecord prepareStructRecord(
             .build();
 
         final Struct key = new Struct(keySchema)
-            .put("id", "123");
+            .put("id", "A123");
         keyChanges.accept(key);
 
         final Struct after = new Struct(valueSchema.field("after").schema())
