fix(dataformat): ignore empty XML elements that contain only whitespaces

Author: fhussonnois

connect-file-pulse-dataformat/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLNodeToStructConverter.java

@@ -25,6 +25,7 @@
 import io.streamthoughts.kafka.connect.filepulse.data.TypedField;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
+import io.streamthoughts.kafka.connect.filepulse.internal.StringUtils;
 import io.streamthoughts.kafka.connect.filepulse.reader.ReaderException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -161,7 +162,7 @@ private TypedValue convertObjectTree(final Node node,
         TypedStruct container = TypedStruct.create();
         getNotExcludedNodeAttributes(node).forEach(container::put);
         for (Node child = node.getFirstChild(); child != null; child = child.getNextSibling()) {
-            // Text nodes always return #text" as the node name, so it's best to use the parent node name instead.
+            // Text nodes always return #text as the node name, so it's best to use the parent node name instead.
             final String childNodeName = isTextNode(child) ? nodeName : determineNodeName(child);
             Optional<TypedValue> optional = readObjectNodeValue(
                     child,
@@ -309,10 +310,10 @@ private static Optional<String> peekChildCDataNodeTextValue(final Node node) {
 
         if (nonNewLineNodes.size() == 1) {
             final Node child = nonNewLineNodes.get(0);
-            if (isTextNode(child)) {
-                // Text content can be an empty string.
+            if (isWhitespaceOrNewLineNodeElement(child))
+                return Optional.empty();
+            if (isTextNode(child))
                 return Optional.of(child.getTextContent());
-            }
         }
 
         return Optional.empty();
@@ -347,7 +348,7 @@ private static boolean isNotXmlNamespace(final Node node) {
     }
 
     private static boolean isWhitespaceOrNewLineNodeElement(final Node node) {
-        return node != null && isTextNode(node) && node.getTextContent().trim().isEmpty();
+        return node != null && isTextNode(node) && StringUtils.isBlank(node.getTextContent());
     }
 
     private String determineNodeName(final Node node) {

connect-file-pulse-dataformat/src/test/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLNodeToStructConverterTest.java

@@ -132,4 +132,61 @@ public void should_convert_given_single_text_node_element_with_attrs() throws Ex
         Assert.assertEquals("test", root.getString("text"));
         Assert.assertEquals("attr", root.getString("attr"));
     }
+
+    @Test
+    public void should_ignore_element_given_xml_tag_with_whitespaces() throws Exception {
+        // Given
+        final byte[] bytes = "<root><empty>            </empty><data>text</data></root>".getBytes();
+        final XMLNodeToStructConverter converter = new XMLNodeToStructConverter()
+                .setContentFieldName("text")
+                .setExcludeEmptyElement(true);
+
+        // When
+        final Document document = reader.parse(new ByteArrayInputStream(bytes));
+        TypedStruct struct = converter.apply(document);
+
+        // Then
+        Assert.assertNotNull(struct);
+        TypedStruct root = struct.getStruct("root");
+        Assert.assertNotNull(root);
+        Assert.assertFalse(root.has("empty"));
+    }
+
+    @Test
+    public void should_ignore_empty_element_given_self_closing_xml_tag() throws Exception {
+        // Given
+        final byte[] bytes = "<root><empty/><data>text</data></root>".getBytes();
+        final XMLNodeToStructConverter converter = new XMLNodeToStructConverter()
+                .setContentFieldName("text")
+                .setExcludeEmptyElement(true);
+
+        // When
+        final Document document = reader.parse(new ByteArrayInputStream(bytes));
+        TypedStruct struct = converter.apply(document);
+
+        // Then
+        Assert.assertNotNull(struct);
+        TypedStruct root = struct.getStruct("root");
+        Assert.assertNotNull(root);
+        Assert.assertFalse(root.has("empty"));
+    }
+
+    @Test
+    public void should_ignore_empty_element_given_open_and_close_xml_tags() throws Exception {
+        // Given
+        final byte[] bytes = "<root><empty></empty><data>text</data></root>".getBytes();
+        final XMLNodeToStructConverter converter = new XMLNodeToStructConverter()
+                .setContentFieldName("text")
+                .setExcludeEmptyElement(true);
+
+        // When
+        final Document document = reader.parse(new ByteArrayInputStream(bytes));
+        TypedStruct struct = converter.apply(document);
+
+        // Then
+        Assert.assertNotNull(struct);
+        TypedStruct root = struct.getStruct("root");
+        Assert.assertNotNull(root);
+        Assert.assertFalse(root.has("empty"));
+    }
 }