Merge pull request #13 from aiven/ivanyu-hashfield-clean-and-readme

Improve Hash implementation and update README

Author: willyborankin

README.md

@@ -44,7 +44,7 @@ Exists in two variants:
 The transformation defines the following configurations:
 
 - `field.name` - The name of the field which should be used as the topic name. If `null` or empty, the entire key or value is used (and assumed to be a string). By default is `null`.
-- `skip.missing.or.null` - In case the source of the new topic name is `null` or missing, should a record be silently passed without transformation. By default is `false`.
+- `skip.missing.or.null` - In case the source of the new topic name is `null` or missing, should a record be silently passed without transformation. By default, is `false`.
 
 Here is an example of this transformation configuration:
 
@@ -54,6 +54,32 @@ transforms.ExtractTopicFromValueField.type=io.aiven.kafka.connect.transforms.Ext
 transforms.ExtractTopicFromValueField.field.name=inner_field_name
 ```
 
+### `Hash`
+
+This transformation replaces a string value with its hash.
+
+The transformation can hash either the whole key or value (in this case, it must have `STRING` type) or a field in them (in this case, it must have `STRUCT` type and the field's value must be `STRING`).
+
+Exists in two variants:
+
+ - `io.aiven.kafka.connect.transforms.Hash$Key` - works on keys;
+ - `io.aiven.kafka.connect.transforms.Hash$Value` - works on values.
+
+The transformation defines the following configurations:
+
+ - `field.name` - 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`.
+ - `function` - The name of the hash function to use. The supported values are: `md5`, `sha1`, and `sha256`.
+ - `skip.missing.or.null` - In case the value to be hashed is `null` or missing, should a record be silently passed without transformation. By default, is `false`.
+
+Here is an example of this transformation configuration:
+
+```
+transforms=HashEmail
+transforms.HashEmail.type=io.aiven.kafka.connect.transforms.Hash$Value
+transforms.HashEmail.field.name=email
+transforms.HashEmail.function=sha1
+```
+
 ### `TombstoneHandler`
 
 This transformation manages tombstone records, 

src/main/java/io/aiven/kafka/connect/transforms/HashField.java renamed to src/main/java/io/aiven/kafka/connect/transforms/Hash.java

@@ -17,9 +17,8 @@
 package io.aiven.kafka.connect.transforms;
 
 import java.security.MessageDigest;
-import java.util.Arrays;
+import java.security.NoSuchAlgorithmException;
 import java.util.Base64;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
@@ -29,29 +28,46 @@
 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.ConnectException;
 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);
+public abstract class Hash<R extends ConnectRecord<R>> implements Transformation<R> {
+    private static final Logger log = LoggerFactory.getLogger(Hash.class);
 
-    private HashFieldConfig config;
-    private static final List<Schema.Type> SUPPORTED_TYPES_TO_CONVERT_FROM = Arrays.asList(
-            Schema.Type.STRING
-    );
+    private HashConfig config;
+    private MessageDigest messageDigest;
 
     @Override
     public ConfigDef config() {
-        return HashFieldConfig.config();
+        return HashConfig.config();
     }
 
     @Override
     public void configure(final Map<String, ?> configs) {
-        this.config = new HashFieldConfig(configs);
+        this.config = new HashConfig(configs);
+
+        try {
+            switch (config.hashFunction()) {
+                case MD5:
+                    messageDigest = MessageDigest.getInstance("MD5");
+                    break;
+                case SHA1:
+                    messageDigest = MessageDigest.getInstance("SHA1");
+                    break;
+                case SHA256:
+                    messageDigest = MessageDigest.getInstance("SHA-256");
+                    break;
+                default:
+                    throw new ConnectException("Unknown hash function " + config.hashFunction());
+            }
+        } catch (final NoSuchAlgorithmException e) {
+            throw new ConnectException(e);
+        }
     }
 
     @Override
@@ -66,12 +82,8 @@ public R apply(final R record) {
             newValue = getNewValueForNamedField(
                     record.toString(), schemaAndValue.schema(), schemaAndValue.value(), config.fieldName().get());
         } else {
-            if (config.skipMissingOrNull()) {
-                newValue = getNewValueWithoutFieldName(
-                        record.toString(), schemaAndValue.schema(), schemaAndValue.value());
-            } else {
-                throw new DataException(dataPlace() + " can't be null or empty: " + record);
-            }
+            newValue = getNewValueWithoutFieldName(
+                    record.toString(), schemaAndValue.schema(), schemaAndValue.value());
         }
 
         if (newValue.isPresent()) {
@@ -114,60 +126,61 @@ private Optional<String> getNewValueForNamedField(final String recordStr,
         final Field field = schema.field(fieldName);
         if (field == null) {
             if (config.skipMissingOrNull()) {
+                log.debug(fieldName + " in " + dataPlace() + " schema is missing, skipping transformation");
                 return Optional.empty();
             } else {
                 throw new DataException(fieldName + " in " + dataPlace() + " schema can't be missing: " + recordStr);
             }
         }
 
-        if (!SUPPORTED_TYPES_TO_CONVERT_FROM.contains(field.schema().type())) {
+        if (field.schema().type() != Schema.Type.STRING) {
             throw new DataException(fieldName + " schema type in " + dataPlace()
-                    + " must be " + SUPPORTED_TYPES_TO_CONVERT_FROM
+                    + " must be " + Schema.Type.STRING
                     + ": " + recordStr);
         }
 
         final Struct struct = (Struct) value;
-
-        final Optional<String> result = Optional.ofNullable(struct.get(fieldName))
-                .map(Object::toString).map(s -> hashString(config.hashFunction(), s));
-
-        if (result.isPresent() && !result.get().equals("")) {
-            return result;
-        } else {
+        final String stringValue = struct.getString(fieldName);
+        if (stringValue == null) {
             if (config.skipMissingOrNull()) {
+                log.debug(fieldName + " in " + dataPlace() + " is null, skipping transformation");
                 return Optional.empty();
             } else {
-                throw new DataException(fieldName + " in " + dataPlace() + " can't be null or empty: " + recordStr);
+                throw new DataException(fieldName + " in " + dataPlace() + " can't be null: " + recordStr);
             }
+        } else {
+            return Optional.of(hashString(stringValue));
         }
     }
 
-    static final String hashString(final MessageDigest md, final String input) {
-        final byte[] digest = md.digest(input.getBytes());
-        return Base64.getEncoder().encodeToString(digest);
-    }
-
     private Optional<String> getNewValueWithoutFieldName(final String recordStr,
                                                          final Schema schema,
                                                          final Object value) {
-        if (!SUPPORTED_TYPES_TO_CONVERT_FROM.contains(schema.type())) {
+        if (schema.type() != Schema.Type.STRING) {
             throw new DataException(dataPlace() + " schema type must be "
-                    + SUPPORTED_TYPES_TO_CONVERT_FROM
+                    + Schema.Type.STRING
                     + " if field name is not specified: "
                     + recordStr);
         }
 
-        final Optional<String> result = Optional.ofNullable(value)
-                .map(Object::toString).map(s -> hashString(config.hashFunction(), s));
-
-        if (result.isPresent() && !result.get().equals("")) {
-            return result;
-        } else {
-            return Optional.empty();
+        if (value == null) {
+            if (config.skipMissingOrNull()) {
+                log.debug(dataPlace() + " is null, skipping transformation");
+                return Optional.empty();
+            } else {
+                throw new DataException(dataPlace() + " can't be null: " + recordStr);
+            }
         }
+
+        return Optional.of(hashString(value.toString()));
+    }
+
+    private String hashString(final String string) {
+        final byte[] digest = messageDigest.digest(string.getBytes());
+        return Base64.getEncoder().encodeToString(digest);
     }
 
-    public static class Key<R extends ConnectRecord<R>> extends HashField<R> {
+    public static class Key<R extends ConnectRecord<R>> extends Hash<R> {
         @Override
         protected SchemaAndValue getSchemaAndValue(final R record) {
             return new SchemaAndValue(record.keySchema(), record.key());
@@ -179,7 +192,7 @@ protected String dataPlace() {
         }
     }
 
-    public static class Value<R extends ConnectRecord<R>> extends HashField<R> {
+    public static class Value<R extends ConnectRecord<R>> extends Hash<R> {
         @Override
         protected SchemaAndValue getSchemaAndValue(final R record) {
             return new SchemaAndValue(record.valueSchema(), record.value());
@@ -191,7 +204,7 @@ protected String dataPlace() {
         }
     }
 
-    protected HashFieldConfig getConfig() {
+    protected HashConfig getConfig() {
         return this.config;
     }
 }

src/main/java/io/aiven/kafka/connect/transforms/HashFieldConfig.java renamed to src/main/java/io/aiven/kafka/connect/transforms/HashConfig.java

@@ -16,32 +16,26 @@
 
 package io.aiven.kafka.connect.transforms;
 
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
 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;
-import org.apache.kafka.connect.errors.DataException;
 
-class HashFieldConfig extends AbstractConfig {
-    public static final String FIELD_NAME_CONFIG = "field.name";
+class HashConfig extends AbstractConfig {
+    private 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 SKIP_MISSING_OR_NULL_CONFIG = "skip.missing.or.null";
+    private static final String SKIP_MISSING_OR_NULL_CONFIG = "skip.missing.or.null";
     private static final String SKIP_MISSING_OR_NULL_DOC =
-            "In case the source of the new topic name is null or missing, "
+            "In case the value to be hashed is null or missing, "
                     + "should a record be silently passed without transformation.";
-    public static final String FUNCTION_CONFIG = "function";
+    private 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.";
-    private MessageDigest md = null;
 
-    HashFieldConfig(final Map<?, ?> originals) {
+    HashConfig(final Map<?, ?> originals) {
         super(config(), originals);
     }
 
@@ -72,20 +66,25 @@ static ConfigDef config() {
     }
 
     public enum HashFunction {
-        MD5,
+        MD5 {
+            public String toString() {
+                return "md5";
+            }
+        },
         SHA1 {
             public String toString() {
-                return "SHA-1";
+                return "sha1";
             }
         },
         SHA256 {
             public String toString() {
-                return "SHA-256";
+                return "sha256";
             }
         };
-        static String stringValues = Stream.of(HashFunction.values())
-                .map(Enum::toString)
-                .collect(Collectors.joining("|", "[", "]"));
+
+        public static HashFunction fromString(final String value) {
+            return valueOf(value.toUpperCase());
+        }
     }
 
     Optional<String> fieldName() {
@@ -100,18 +99,8 @@ boolean skipMissingOrNull() {
         return getBoolean(SKIP_MISSING_OR_NULL_CONFIG);
     }
 
-    MessageDigest hashFunction() {
-        final String hashAlg = getString(FUNCTION_CONFIG);
-        try {
-            if (md == null) {
-                md = MessageDigest.getInstance(hashAlg);
-            }
-        } catch (final NoSuchAlgorithmException e) {
-            throw new DataException("Hash function " + hashAlg
-                    + " must be "
-                    + HashFieldConfig.HashFunction.stringValues);
-        }
-        return md;
+    HashFunction hashFunction() {
+        return HashFunction.fromString(getString(FUNCTION_CONFIG));
     }
 }