Merge pull request #31 from brbrown25/bugfix/fix-broken-hash-transformer

bug(Hash Transformer): Fixing an issue where the hash transformer tes…

Author: ivanyu

build.gradle

@@ -42,7 +42,7 @@ targetCompatibility = JavaVersion.VERSION_11
 
 ext {
     kafkaVersion = "2.0.1"
-    testcontainersVersion = "1.12.1"
+    testcontainersVersion = "1.15.1"
 }
 
 distributions {

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

@@ -71,36 +71,16 @@ public void configure(final Map<String, ?> 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);
-        }
-
+    public final Optional<Object> getNewValue(final R record, final SchemaAndValue schemaAndValue) {
         final Optional<Object> newValue;
         if (config.fieldName().isPresent()) {
             newValue = getNewValueForNamedField(
-                    record.toString(), schemaAndValue.schema(), schemaAndValue.value(), config.fieldName().get());
+                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;
+                record.toString(), schemaAndValue.schema(), schemaAndValue.value());
         }
+        return newValue;
     }
 
     @Override
@@ -190,6 +170,31 @@ protected SchemaAndValue getSchemaAndValue(final R record) {
             return new SchemaAndValue(record.keySchema(), record.key());
         }
 
+        @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<Object> newValue = getNewValue(record, schemaAndValue);
+
+            if (newValue.isPresent()) {
+                return record.newRecord(
+                    record.topic(),
+                    record.kafkaPartition(),
+                    record.keySchema(),
+                    newValue.get(),
+                    record.valueSchema(),
+                    record.value(),
+                    record.timestamp(),
+                    record.headers()
+                );
+            } else {
+                return record;
+            }
+        }
+
         @Override
         protected String dataPlace() {
             return "key";
@@ -202,6 +207,31 @@ protected SchemaAndValue getSchemaAndValue(final R record) {
             return new SchemaAndValue(record.valueSchema(), record.value());
         }
 
+        @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<Object> newValue = getNewValue(record, schemaAndValue);
+
+            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
         protected String dataPlace() {
             return "value";

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

@@ -35,4 +35,17 @@ protected Hash<SinkRecord> createTransformationObject() {
     protected SinkRecord record(final Schema schema, final Object data) {
         return record(schema, data, null, null);
     }
+
+    @Override
+    protected SinkRecord setNewValue(final SinkRecord record, final Object newValue) {
+        return record.newRecord(record.topic(),
+            record.kafkaPartition(),
+            record.keySchema(),
+            newValue,
+            record.valueSchema(),
+            record.value(),
+            record.timestamp(),
+            record.headers()
+        );
+    }
 }

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

@@ -287,6 +287,8 @@ private Hash<SinkRecord> transformation(
 
     protected abstract Hash<SinkRecord> createTransformationObject();
 
+    protected abstract SinkRecord setNewValue(final SinkRecord record, final Object newValue);
+
     protected abstract SinkRecord record(final Schema schema, final Object data);
 
     protected SinkRecord record(final Schema keySchema,
@@ -300,18 +302,6 @@ protected SinkRecord record(final Schema keySchema,
                 456L, TimestampType.CREATE_TIME);
     }
 
-    private SinkRecord setNewValue(final SinkRecord record, final Object newValue) {
-        return record.newRecord(record.topic(),
-                record.kafkaPartition(),
-                record.keySchema(),
-                record.key(),
-                record.valueSchema(),
-                newValue,
-                record.timestamp(),
-                record.headers()
-        );
-    }
-
     private String hash(final String function, final String value) {
         return HASHED_VALUES.get(function).get(value);
     }

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

@@ -35,4 +35,17 @@ protected Hash<SinkRecord> createTransformationObject() {
     protected SinkRecord record(final Schema schema, final Object data) {
         return record(null, null, schema, data);
     }
+
+    @Override
+    protected SinkRecord setNewValue(final SinkRecord record, final Object newValue) {
+        return record.newRecord(record.topic(),
+            record.kafkaPartition(),
+            record.keySchema(),
+            record.key(),
+            record.valueSchema(),
+            newValue,
+            record.timestamp(),
+            record.headers()
+        );
+    }
 }