fix(api): connector should throw an error when merging struct with string

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/schema/SchemaMerger.java

@@ -77,7 +77,8 @@ public static Schema merge(final Schema left, final Schema right, final SchemaCo
             return left;
         }
 
-        if (left.type() == Type.STRING || right.type() == Type.STRING)
+        if ( (left.type() == Type.STRING && right.type().isPrimitive()) ||
+             (right.type() == Type.STRING && left.type().isPrimitive()) )
             return mergeMetadata(left, right, SchemaBuilder.string());
 
         if ( (left.type() == Type.INT64 && right.type() == Type.INT32) ||

connect-file-pulse-api/src/test/java/io/streamthoughts/kafka/connect/filepulse/schema/SchemaMergerTest.java

@@ -18,6 +18,7 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.schema;
 
+import io.streamthoughts.kafka.connect.filepulse.data.DataException;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.junit.Assert;
@@ -168,4 +169,15 @@ public void should_success_merge_given_duplicate_schemas() {
                 schema.field(DEFAULT_FIELD_C).schema().field("field3").schema()
         );
     }
+
+    @Test(expected = DataException.class)
+    public void should_throw_error_when_merging_struct_given_string() {
+        // Given
+        Schema schemaLeft = SchemaBuilder.struct()
+                .field(DEFAULT_FIELD_A, SchemaBuilder.array(SchemaBuilder.string()))
+                .build();
+
+        // When
+        SchemaMerger.merge(schemaLeft, SchemaBuilder.string());
+    }
 }