fix(api): fix regression on DefaultTypeValueMerger

streamthoughts · Sep 10, 2020 · 6ec42eb · 6ec42eb
1 parent 99288ff
commit 6ec42eb
Show file tree

Hide file tree

Showing 2 changed files with 75 additions and 11 deletions.
diff --git a/...in/java/io/streamthoughts/kafka/connect/filepulse/data/merger/DefaultTypeValueMerger.java b/...in/java/io/streamthoughts/kafka/connect/filepulse/data/merger/DefaultTypeValueMerger.java
@@ -86,13 +86,13 @@ private static TypedStruct mergeObjects(final TypedStruct left,
                 final TypedStruct mergedStruct = mergeObjects(
                     leftValue.getStruct(),
                     rightValue.getStruct(),
-                    overwrite.stream().map(Path::forward).filter(Objects::nonNull).collect(Collectors.toSet())
+                    computeNextOverwrite(overwrite, fieldName)
                 );
                 merged = TypedValue.struct(mergedStruct);
             } else {
                 merged = merge(leftValue, rightValue);
             }
-            struct.put(leftField.name(), merged);
+            struct.put(fieldName, merged);
         }
 
         for (TypedField f : right) {
@@ -103,6 +103,13 @@ private static TypedStruct mergeObjects(final TypedStruct left,
         return struct;
     }
 
+    private static Set<Path> computeNextOverwrite(final Set<Path> overwrite, final String fieldName) {
+        return overwrite.stream()
+            .map(p -> p.forwardIfOrNull(fieldName))
+            .filter(Objects::nonNull)
+            .collect(Collectors.toSet());
+    }
+
     private static class Path {
 
         private final String path;
@@ -122,11 +129,15 @@ private static class Path {
         }
 
         public boolean matches(final String field) {
-            return this.field.equals(field) || path.equals(field);
+            return path.equals(field);
         }
 
-        private Path forward() {
-            return remaining == null ? null : new Path(remaining);
+        private Path forwardIfOrNull(final String field) {
+            if (!this.field.equals(field))
+                return null;
+            if (remaining == null)
+                return null;
+            return new Path(remaining);
         }
     }
 

diff --git a/.../test/java/io/streamthoughts/kafka/connect/filepulse/data/merger/TypeValueMergerTest.java b/.../test/java/io/streamthoughts/kafka/connect/filepulse/data/merger/TypeValueMergerTest.java
@@ -36,7 +36,7 @@ public class TypeValueMergerTest {
     private final TypeValueMerger merger = new DefaultTypeValueMerger();
 
     @Test
-    public void shouldMergeStructGivenTwoFieldsWithDifferentName() {
+    public void should_merge_struct_given_two_fields_with_different_name() {
         final TypedStruct structLeft = TypedStruct.create()
                 .put(FIELD_VALUE_A, VALUE_A);
 
@@ -52,7 +52,7 @@ public void shouldMergeStructGivenTwoFieldsWithDifferentName() {
     }
 
     @Test
-    public void shouldMergeStructGivenTwoFieldsWithDifferentTypeGivenOverride() {
+    public void should_merge_struct_given_two_fields_with_different_type_given_override() {
         final TypedStruct structLeft = TypedStruct.create()
                 .put(FIELD_VALUE_A, VALUE_A);
 
@@ -66,7 +66,7 @@ public void shouldMergeStructGivenTwoFieldsWithDifferentTypeGivenOverride() {
     }
 
     @Test
-    public void shouldMergeStructGivenTwoFieldsWithSameNameIntoArray() {
+    public void should_merge_struct_given_two_fields_with_same_name_into_array() {
         final TypedStruct structLeft = TypedStruct.create()
                 .put(FIELD_VALUE_A, VALUE_A);
 
@@ -83,7 +83,60 @@ public void shouldMergeStructGivenTwoFieldsWithSameNameIntoArray() {
     }
 
     @Test
-    public void shouldMergeStructGivenLeftFieldWithArrayTypeEqualToRightField() {
+    public void should_merge_struct_given_two_identical_paths_into_array() {
+        final TypedStruct structLeft = TypedStruct.create()
+                .insert("a.b1", VALUE_A)
+                .insert("a.b2", VALUE_A);
+
+        final TypedStruct structRight = TypedStruct.create()
+                .insert("a.b2", VALUE_B);
+
+        final TypedStruct merged = merger.merge(structLeft, structRight, Collections.emptySet());
+
+        assertNotNull(merged);
+
+        assertEquals(VALUE_A,  merged.find("a.b1").getString());
+        assertEquals(2,  merged.find("a.b2").getArray().size());
+        assertTrue(merged.find("a.b2").getArray().contains(VALUE_A));
+        assertTrue(merged.find("a.b2").getArray().contains(VALUE_B));
+    }
+
+    @Test
+    public void should_merge_struct_given_two_existing_paths_and_overwrite() {
+        final TypedStruct structLeft = TypedStruct.create()
+                .insert("a.b1", VALUE_A)
+                .insert("a.b2", VALUE_A);
+
+        final TypedStruct structRight = TypedStruct.create()
+                .insert("a.b2", VALUE_B);
+
+        final TypedStruct merged = merger.merge(structLeft, structRight, Collections.singleton("a.b2"));
+
+        assertNotNull(merged);
+
+        assertEquals(VALUE_A,  merged.find("a.b1").getString());
+        assertEquals(VALUE_B,  merged.find("a.b2").getString());
+    }
+
+    @Test
+    public void should_merge_struct_given_two_existing_paths_into_array() {
+        final TypedStruct structLeft = TypedStruct.create()
+                .insert("a.b1", VALUE_A)
+                .insert("a.b2", VALUE_A);
+
+        final TypedStruct structRight = TypedStruct.create()
+                .insert("a.b2", VALUE_B);
+
+        final TypedStruct merged = merger.merge(structLeft, structRight, Collections.emptySet());
+
+        assertNotNull(merged);
+
+        assertEquals(VALUE_A,  merged.find("a.b1").getString());
+        assertEquals(2,  merged.find("a.b2").getArray().size());
+    }
+
+    @Test
+    public void should_merge_struct_given_left_field_with_array_type_equal_to_right_field() {
         final TypedStruct structLeft = TypedStruct.create()
                 .put(FIELD_VALUE_A, Collections.singletonList(VALUE_A));
 
@@ -117,7 +170,7 @@ public void shouldMergeStructGivenRightFieldWithArrayTypeEqualToLeftField() {
     }
 
     @Test
-    public void shouldMergeStructGivenTwoArrayFieldsWithEqualsValueType() {
+    public void should_merge_struct_given_two_array_fields_with_equals_value_type() {
         final TypedStruct structLeft = TypedStruct.create()
                 .put(FIELD_VALUE_A, Collections.singletonList(VALUE_A));
 
@@ -134,7 +187,7 @@ public void shouldMergeStructGivenTwoArrayFieldsWithEqualsValueType() {
     }
 
     @Test
-    public void shouldMergedGivenTwoStructWithCommonChildFields() {
+    public void should_merged_given_two_struct_with_common_child_fields() {
         final TypedStruct structLeft = TypedStruct.create()
             .insert("a.b1.c", VALUE_A);