feat(HashField): Adding a HashField transformer that can use a supplied hashing algorithm to transform the targeted value. ISSUE-9.

Author: Brandon Brown

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

@@ -0,0 +1,192 @@
+/*
+ * Copyright 2019 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.connect.transforms;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.config.ConfigDef;
+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.errors.DataException;
+import org.apache.kafka.connect.transforms.Transformation;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public abstract class HashField<R extends ConnectRecord<R>> implements Transformation<R> {
+    private static final Logger log = LoggerFactory.getLogger(HashField.class);
+
+    private HashFieldConfig config;
+    private static final List<Schema.Type> SUPPORTED_TYPES_TO_CONVERT_FROM = Arrays.asList(
+            Schema.Type.STRING
+    );
+
+    @Override
+    public ConfigDef config() {
+        return HashFieldConfig.config();
+    }
+
+    @Override
+    public void configure(final Map<String, ?> configs) {
+        this.config = new HashFieldConfig(configs);
+    }
+
+    @Override
+    public R apply(final R record) {
+        final SchemaAndValue schemaAndValue = getSchemaAndValue(record);
+        if (schemaAndValue.schema() == null) {
+            throw new DataException(dataPlace() + " schema can't be null: " + record);
+        }
+
+        final Optional<String> newValue;
+        if (config.fieldName().isPresent()) {
+            newValue = getNewValueForNamedField(
+                    record.toString(), schemaAndValue.schema(), schemaAndValue.value(), config.fieldName().get());
+        } else {
+            newValue = getNewValueWithoutFieldName(
+                    record.toString(), schemaAndValue.schema(), schemaAndValue.value());
+        }
+
+        if (newValue.isPresent()) {
+            return record.newRecord(
+                    record.topic(),
+                    record.kafkaPartition(),
+                    record.keySchema(),
+                    record.key(),
+                    record.valueSchema(),
+                    newValue.get(),
+                    record.timestamp(),
+                    record.headers()
+            );
+        } else {
+            return record;
+        }
+    }
+
+    @Override
+    public void close() {
+    }
+
+    protected abstract String dataPlace();
+
+    protected abstract SchemaAndValue getSchemaAndValue(final R record);
+
+    private Optional<String> getNewValueForNamedField(final String recordStr,
+                                                      final Schema schema,
+                                                      final Object value,
+                                                      final String fieldName) {
+        if (Schema.Type.STRUCT != schema.type()) {
+            throw new DataException(dataPlace() + " schema type must be STRUCT if field name is specified: "
+                    + recordStr);
+        }
+
+        if (value == null) {
+            throw new DataException(dataPlace() + " can't be null if field name is specified: " + recordStr);
+        }
+
+        final Field field = schema.field(fieldName);
+        if (field == null) {
+            return Optional.empty();
+        }
+
+        if (!SUPPORTED_TYPES_TO_CONVERT_FROM.contains(field.schema().type())) {
+            throw new DataException(fieldName + " schema type in " + dataPlace()
+                    + " must be " + SUPPORTED_TYPES_TO_CONVERT_FROM
+                    + ": " + recordStr);
+        }
+
+        final Struct struct = (Struct) value;
+
+        final Optional<String> result = Optional.ofNullable(struct.get(fieldName))
+                .map(Object::toString).flatMap(s -> config.hashFunction().map(alg -> hashString(alg, s)));
+        if (result.isPresent() && !result.get().equals("")) {
+            return result;
+        } else {
+            return Optional.empty();
+        }
+    }
+
+    static final String hashString(final String hashAlg, final String input) {
+        try {
+            final MessageDigest md = MessageDigest.getInstance(hashAlg);
+            final byte[] digest = md.digest(input.getBytes());
+            return Base64.getEncoder().encodeToString(digest);
+        } catch (final NoSuchAlgorithmException e) {
+            System.out.println("EXCEPTION: " + e.getMessage());
+            throw new DataException("Hash function " + hashAlg
+                    + " must be "
+                    + HashFieldConfig.HashFunction.stringValues
+                    .stream()
+                    .collect(Collectors.joining("|", "[", "]")));
+        }
+    }
+
+    private Optional<String> getNewValueWithoutFieldName(final String recordStr,
+                                                         final Schema schema,
+                                                         final Object value) {
+        if (!SUPPORTED_TYPES_TO_CONVERT_FROM.contains(schema.type())) {
+            throw new DataException(dataPlace() + " schema type must be "
+                    + SUPPORTED_TYPES_TO_CONVERT_FROM
+                    + " if field name is not specified: "
+                    + recordStr);
+        }
+
+        final Optional<String> result = Optional.ofNullable(value)
+                .map(Object::toString).flatMap(s -> config.hashFunction().map(alg -> hashString(alg, s)));
+
+        if (result.isPresent() && !result.get().equals("")) {
+            return result;
+        } else {
+            return Optional.empty();
+        }
+    }
+
+    public static class Key<R extends ConnectRecord<R>> extends HashField<R> {
+        @Override
+        protected SchemaAndValue getSchemaAndValue(final R record) {
+            return new SchemaAndValue(record.keySchema(), record.key());
+        }
+
+        @Override
+        protected String dataPlace() {
+            return "key";
+        }
+    }
+
+    public static class Value<R extends ConnectRecord<R>> extends HashField<R> {
+        @Override
+        protected SchemaAndValue getSchemaAndValue(final R record) {
+            return new SchemaAndValue(record.valueSchema(), record.value());
+        }
+
+        @Override
+        protected String dataPlace() {
+            return "value";
+        }
+    }
+}

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

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2019 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.connect.transforms;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+class HashFieldConfig extends AbstractConfig {
+    public static final String FIELD_NAME_CONFIG = "field.name";
+    private static final String FIELD_NAME_DOC =
+            "The name of the field which value should be hashed. If null or empty, "
+                    + "the entire key or value is used (and assumed to be a string). By default is null.";
+    public static final String FUNCTION_CONFIG = "function";
+    private static final String FUNCTION_DOC =
+            "The name of the hash function to use. The supported values are: md5, sha1, sha256.";
+
+    HashFieldConfig(final Map<?, ?> originals) {
+        super(config(), originals);
+    }
+
+    static ConfigDef config() {
+        return new ConfigDef()
+                .define(
+                        FIELD_NAME_CONFIG,
+                        ConfigDef.Type.STRING,
+                        null,
+                        ConfigDef.Importance.HIGH,
+                        FIELD_NAME_DOC)
+                .define(
+                        FUNCTION_CONFIG,
+                        ConfigDef.Type.STRING,
+                        null,
+                        HashFunctionValidator.INSTANCE,
+                        ConfigDef.Importance.HIGH,
+                        FUNCTION_DOC);
+    }
+
+    public enum HashFunction {
+        MD5,
+        SHA1 {
+            public String toString() {
+                return "SHA-1";
+            }
+        },
+        SHA256 {
+            public String toString() {
+                return "SHA-256";
+            }
+        };
+        static List<String> stringValues = Stream.of(HashFunction.values())
+                .map(Enum::toString)
+                .collect(Collectors.toList());
+    }
+
+    Optional<String> fieldName() {
+        final String rawFieldName = getString(FIELD_NAME_CONFIG);
+        if (null == rawFieldName || "".equals(rawFieldName)) {
+            return Optional.empty();
+        }
+        return Optional.of(rawFieldName);
+    }
+
+    Optional<String> hashFunction() {
+        final String rawFieldName = getString(FUNCTION_CONFIG);
+        if (null == rawFieldName || "".equals(rawFieldName)) {
+            return Optional.empty();
+        }
+        return Optional.of(rawFieldName);
+    }
+}
+

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2019 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.connect.transforms;
+
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+
+public class HashFunctionValidator implements ConfigDef.Validator {
+    public static final HashFunctionValidator INSTANCE =
+            new HashFunctionValidator();
+
+    @Override
+    public void ensureValid(final String name, final Object value) {
+        if (value != null) {
+            if (!HashFieldConfig.HashFunction.stringValues.contains(value.toString())) {
+                throw new ConfigException(name, value, "Invalid Hash Function");
+            }
+        }
+    }
+
+    @Override
+    public String toString() {
+        final String validFunctions = HashFieldConfig.HashFunction.stringValues
+                .stream()
+                .collect(Collectors.joining("|"));
+        return "Must be one of " + validFunctions;
+    }
+}

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2019 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.connect.transforms;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class HashFieldConfigTest {
+    @Test
+    void defaults() {
+        final Map<String, String> props = new HashMap<>();
+        final HashFieldConfig config = new HashFieldConfig(props);
+        assertEquals(Optional.empty(), config.fieldName());
+        assertEquals(Optional.empty(), config.hashFunction());
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"MD5", "SHA-1", "SHA-256"})
+    void hashFunction(final String hashFunction) {
+        final Map<String, String> props = new HashMap<>();
+        props.put(HashFieldConfig.FUNCTION_CONFIG, hashFunction);
+        final HashFieldConfig config = new HashFieldConfig(props);
+        assertEquals(Optional.of(hashFunction), config.hashFunction());
+    }
+
+    @Test
+    void emptyFieldName() {
+        final Map<String, String> props = new HashMap<>();
+        props.put(HashFieldConfig.FIELD_NAME_CONFIG, "");
+        final HashFieldConfig config = new HashFieldConfig(props);
+        assertEquals(Optional.empty(), config.fieldName());
+    }
+
+    @Test
+    void definedFieldName() {
+        final Map<String, String> props = new HashMap<>();
+        props.put(HashFieldConfig.FIELD_NAME_CONFIG, "test");
+        final HashFieldConfig config = new HashFieldConfig(props);
+        assertEquals(Optional.of("test"), config.fieldName());
+    }
+}