Merge pull request #17 from brbrown25/bugfix/ISSUE-15

FIX: Hash hashing whole structure when field name set

Author: ivanyu

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

@@ -77,7 +77,7 @@ public R apply(final R record) {
             throw new DataException(dataPlace() + " schema can't be null: " + record);
         }
 
-        final Optional<String> newValue;
+        final Optional<Object> newValue;
         if (config.fieldName().isPresent()) {
             newValue = getNewValueForNamedField(
                     record.toString(), schemaAndValue.schema(), schemaAndValue.value(), config.fieldName().get());
@@ -110,7 +110,7 @@ public void close() {
 
     protected abstract SchemaAndValue getSchemaAndValue(final R record);
 
-    private Optional<String> getNewValueForNamedField(final String recordStr,
+    private Optional<Object> getNewValueForNamedField(final String recordStr,
                                                       final Schema schema,
                                                       final Object value,
                                                       final String fieldName) {
@@ -149,11 +149,13 @@ private Optional<String> getNewValueForNamedField(final String recordStr,
                 throw new DataException(fieldName + " in " + dataPlace() + " can't be null: " + recordStr);
             }
         } else {
-            return Optional.of(hashString(stringValue));
+            final String updatedValue = hashString(stringValue);
+            final Struct updatedRecord = struct.put(fieldName, updatedValue);
+            return Optional.ofNullable(updatedRecord);
         }
     }
 
-    private Optional<String> getNewValueWithoutFieldName(final String recordStr,
+    private Optional<Object> getNewValueWithoutFieldName(final String recordStr,
                                                          final Schema schema,
                                                          final Object value) {
         if (schema.type() != Schema.Type.STRING) {

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

@@ -42,6 +42,8 @@ abstract class HashTest {
     private static final String EMPTY_FIELD_VALUE = "";
     private static final String NON_EMPTY_FIELD_VALUE = "jerry@all_your_bases.com";
     private static final String DEFAULT_HASH_FUNCTION = HashConfig.HashFunction.SHA256.toString();
+    private static final String UNAFFECTED_FIELD = "name";
+    private static final String UNAFFECTED_FIELD_VALUE = "jerry";
 
     @Test
     void noFieldName_NullValue_NoSkip() {
@@ -132,8 +134,12 @@ void fieldName_MissingValue_NoSkip() {
     void fieldName_NullValue_Skip() {
         final Schema schema = SchemaBuilder.struct()
                 .field(FIELD, SchemaBuilder.OPTIONAL_STRING_SCHEMA)
+                .field(UNAFFECTED_FIELD, SchemaBuilder.STRING_SCHEMA)
                 .schema();
-        final SinkRecord originalRecord = record(schema, new Struct(schema).put(FIELD, null));
+        final Struct originalStruct = new Struct(schema)
+                .put(FIELD, null)
+                .put(UNAFFECTED_FIELD, UNAFFECTED_FIELD_VALUE);
+        final SinkRecord originalRecord = record(schema, originalStruct);
         final Hash<SinkRecord> transform = transformation(FIELD, true, DEFAULT_HASH_FUNCTION);
         final SinkRecord result = transform.apply(originalRecord);
         // No changes.
@@ -193,11 +199,17 @@ void fieldName_UnsupportedTypeInField() {
     void fieldName_NormalStringValue(final String hashFunction) {
         final Schema schema = SchemaBuilder.struct()
                 .field(FIELD, SchemaBuilder.STRING_SCHEMA)
+                .field(UNAFFECTED_FIELD, SchemaBuilder.STRING_SCHEMA)
                 .schema();
-        final SinkRecord originalRecord = record(schema, new Struct(schema).put(FIELD, NON_EMPTY_FIELD_VALUE));
+        final Struct originalStruct = new Struct(schema)
+                .put(FIELD, NON_EMPTY_FIELD_VALUE)
+                .put(UNAFFECTED_FIELD, UNAFFECTED_FIELD_VALUE);
+        final SinkRecord originalRecord = record(schema, originalStruct);
         final Hash<SinkRecord> transform = transformation(FIELD, true, hashFunction);
         final SinkRecord result = transform.apply(originalRecord);
-        final String newValue = hash(hashFunction, NON_EMPTY_FIELD_VALUE);
+        final Struct newValue = new Struct(schema)
+                .put(FIELD, hash(hashFunction, NON_EMPTY_FIELD_VALUE))
+                .put(UNAFFECTED_FIELD, UNAFFECTED_FIELD_VALUE);
         assertEquals(setNewValue(originalRecord, newValue), result);
     }
 
@@ -206,11 +218,17 @@ void fieldName_NormalStringValue(final String hashFunction) {
     void fieldName_EmptyStringValue(final String hashFunction) {
         final Schema schema = SchemaBuilder.struct()
                 .field(FIELD, SchemaBuilder.STRING_SCHEMA)
+                .field(UNAFFECTED_FIELD, SchemaBuilder.STRING_SCHEMA)
                 .schema();
-        final SinkRecord originalRecord = record(schema, new Struct(schema).put(FIELD, EMPTY_FIELD_VALUE));
+        final Struct originalStruct = new Struct(schema)
+                .put(FIELD, EMPTY_FIELD_VALUE)
+                .put(UNAFFECTED_FIELD, UNAFFECTED_FIELD_VALUE);
+        final SinkRecord originalRecord = record(schema, originalStruct);
         final Hash<SinkRecord> transform = transformation(FIELD, true, hashFunction);
         final SinkRecord result = transform.apply(originalRecord);
-        final String newValue = hash(hashFunction, EMPTY_FIELD_VALUE);
+        final Struct newValue = new Struct(schema)
+                .put(FIELD, hash(hashFunction, EMPTY_FIELD_VALUE))
+                .put(UNAFFECTED_FIELD, UNAFFECTED_FIELD_VALUE);
         assertEquals(setNewValue(originalRecord, newValue), result);
     }
 
@@ -246,7 +264,7 @@ protected SinkRecord record(final Schema keySchema,
                 456L, TimestampType.CREATE_TIME);
     }
 
-    private SinkRecord setNewValue(final SinkRecord record, final String newValue) {
+    private SinkRecord setNewValue(final SinkRecord record, final Object newValue) {
         return record.newRecord(record.topic(),
                 record.kafkaPartition(),
                 record.keySchema(),