fix(plugins): fix XMLFileInputReader on the root document when force.array.on.fields is used (#75)

Resolves: #75

Author: fhussonnois

connect-file-pulse-plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/reader/XMLFileInputReader.java

@@ -215,16 +215,19 @@ private static class Node2StructConverter {
          */
         private static TypedStruct convertNodeObjectTree(final Node node, final FieldPaths forceArrayFields) {
             Objects.requireNonNull(node, "node cannot be null");
-            final FieldPaths currentForceArrayFields = forceArrayFields.next(determineNodeName(node));
+            String nodeName = determineNodeName(node);
+            final FieldPaths currentForceArrayFields = nodeName.equals("#document")
+                    ? forceArrayFields :
+                    forceArrayFields.next(nodeName);
             TypedStruct container = TypedStruct.create();
             addAllNodeAttributes(container, node.getAttributes());
             for (Node child = node.getFirstChild(); child != null; child = child.getNextSibling()) {
-                final String nodeName = isTextNode(child) ? determineNodeName(node) : determineNodeName(child);
+                final String childNodeName = isTextNode(child) ? nodeName : determineNodeName(child);
                 Optional<?> optional = readNodeObject(child, currentForceArrayFields);
                 if (optional.isPresent()) {
                     Object nodeValue = optional.get();
-                    final boolean isArray = currentForceArrayFields.anyMatches(nodeName);
-                    container = enrichStructWithObject(container, nodeName, nodeValue, isArray);
+                    final boolean isArray = currentForceArrayFields.anyMatches(childNodeName);
+                    container = enrichStructWithObject(container, childNodeName, nodeValue, isArray);
 
                 }
             }

connect-file-pulse-plugin/src/test/java/io/streamthoughts/kafka/connect/filepulse/reader/XMLFileInputReaderTest.java

@@ -183,14 +183,30 @@ public void should_read_record_given_valid_force_array_fields() {
         iterator.forEachRemaining(r -> records.addAll(r.collect()));
 
         Assert.assertEquals(1, records.size());
-        TypedStruct broker = records.get(0).value();
-        Assert.assertEquals(Type.ARRAY, broker.get("topicPartition").type());
-        Assert.assertEquals(1, broker.get("topicPartition").getArray().size());
-        TypedValue segment = ((TypedStruct) broker.get("topicPartition").getArray().iterator().next()).find("numSegments");
+
+        TypedStruct record = records.get(0).value();
+        Assert.assertEquals(Type.ARRAY, record.get("topicPartition").type());
+        Assert.assertEquals(1, record.get("topicPartition").getArray().size());
+        TypedValue segment = ((TypedStruct) record.get("topicPartition").getArray().iterator().next()).find("numSegments");
         Assert.assertEquals(Type.ARRAY, segment.type());
         Assert.assertEquals(1, segment.getArray().size());
     }
 
+    @Test
+    public void should_read_record_given_valid_force_array_fields_and_default_xpath() {
+        reader.configure(new HashMap<String, String>(){{
+            put(FORCE_ARRAY_ON_FIELDS_CONFIG, "cluster.broker.topicPartition");
+        }});
+
+        FileInputIterator<FileRecord<TypedStruct>> iterator = reader.newIterator(context);
+        List<FileRecord<TypedStruct>> records = new ArrayList<>();
+        iterator.forEachRemaining(r -> records.addAll(r.collect()));
+
+        TypedStruct record = records.get(0).value();
+        TypedStruct element = (TypedStruct)record.find("cluster.broker").getArray().iterator().next();
+        Assert.assertEquals(Type.ARRAY, element.find("topicPartition").type());
+    }
+
     @Test
     public void should_read_record_given_single_text_node_with_attrs() throws IOException {
         try(XMLFileInputReader reader = createNewXMLFileInputReader(TEXT_NODE_TEST_XML_DOCUMENT)) {
@@ -205,7 +221,6 @@ public void should_read_record_given_single_text_node_with_attrs() throws IOExce
         }
     }
 
-
     private XMLFileInputReader createNewXMLFileInputReader(final String xmlDocument) throws IOException {
         File file = testFolder.newFile();
         try (BufferedWriter bw = Files.newBufferedWriter(file.toPath(), Charset.defaultCharset())) {